Add stream upstream, stream server zones metrics support

Makefile

@@ -7,7 +7,7 @@ docker-build:
 	docker build --build-arg NGINX_PLUS_VERSION=$(NGINX_PLUS_VERSION)~stretch -t $(NGINX_IMAGE) docker
 
 run-nginx-plus:
-	docker run -d --name nginx-plus-test --rm -p 8080:8080 $(NGINX_IMAGE)
+	docker run -d --name nginx-plus-test --rm -p 8080:8080 -p 8081:8081 $(NGINX_IMAGE)
 
 test-run:
 	go test client/*

client/nginx.go

@@ -60,11 +60,13 @@ type apiError struct {
 // Stats represents NGINX Plus stats fetched from the NGINX Plus API.
 // https://nginx.org/en/docs/http/ngx_http_api_module.html
 type Stats struct {
-	Connections  Connections
-	HTTPRequests HTTPRequests
-	SSL          SSL
-	ServerZones  ServerZones
-	Upstreams    Upstreams
+	Connections       Connections
+	HTTPRequests      HTTPRequests
+	SSL               SSL
+	ServerZones       ServerZones
+	Upstreams         Upstreams
+	StreamServerZones StreamServerZones
+	StreamUpstreams   StreamUpstreams
 }
 
 // Connections represents connection related stats.
@@ -101,7 +103,20 @@ type ServerZone struct {
 	Sent       uint64
 }
 
-// Responses represents HTTP reponse related stats.
+// StreamServerZones is map of stream server zone stats by zone name.
+type StreamServerZones map[string]StreamServerZone
+
+// StreamServerZone represents stream server zone related stats.
+type StreamServerZone struct {
+	Processing  uint64
+	Connections uint64
+	Sessions    Sessions
+	Discarded   uint64
+	Received    uint64
+	Sent        uint64
+}
+
+// Responses represents HTTP response related stats.
 type Responses struct {
 	Responses1xx uint64 `json:"1xx"`
 	Responses2xx uint64 `json:"2xx"`
@@ -111,6 +126,14 @@ type Responses struct {
 	Total        uint64
 }
 
+// Sessions represents stream session related stats.
+type Sessions struct {
+	Sessions2xx uint64 `json:"2xx"`
+	Sessions4xx uint64 `josn:"4xx"`
+	Sessions5xx uint64 `josn:"5xx"`
+	Total       uint64
+}
+
 // Upstreams is a map of upstream stats by upstream name.
 type Upstreams map[string]Upstream
 
@@ -123,6 +146,16 @@ type Upstream struct {
 	Queue      Queue
 }
 
+// StreamUpstreams is a map of stream upstream stats by upstream name.
+type StreamUpstreams map[string]StreamUpstream
+
+// StreamUpstream represents stream upstream related stats.
+type StreamUpstream struct {
+	Peers   []StreamPeer
+	Zombies int
+	Zone    string
+}
+
 // Queue represents queue related stats for an upstream.
 type Queue struct {
 	Size      int
@@ -155,6 +188,31 @@ type Peer struct {
 	ResponseTime uint64 `json:"response_time"`
 }
 
+// StreamPeer represents peer (stream upstream server) related stats.
+type StreamPeer struct {
+	ID            int
+	Server        string
+	Service       string
+	Name          string
+	Backup        bool
+	Weight        int
+	State         string
+	Active        uint64
+	MaxConns      int `json:"max_conns"`
+	Connections   uint64
+	ConnectTime   int    `json:"connect_time"`
+	FirstByteTime int    `json:"first_byte_time"`
+	ResponseTime  uint64 `json:"response_time"`
+	Sent          uint64
+	Received      uint64
+	Fails         uint64
+	Unavail       uint64
+	HealthChecks  HealthChecks `json:"health_checks"`
+	Downtime      uint64
+	Downstart     string
+	Selected      string
+}
+
 // HealthChecks represents health check related stats for a peer.
 type HealthChecks struct {
 	Checks     uint64
@@ -458,7 +516,7 @@ func (client *NginxClient) GetStreamServers(upstream string) ([]StreamUpstreamSe
 	return servers, nil
 }
 
-// AddStreamServer adds the server to the upstream.
+// AddStreamServer adds the stream server to the upstream.
 func (client *NginxClient) AddStreamServer(upstream string, server StreamUpstreamServer) error {
 	id, err := client.getIDOfStreamServer(upstream, server.Server)
 
@@ -572,7 +630,7 @@ func determineStreamUpdates(updatedServers []StreamUpstreamServer, nginxServers
 	return
 }
 
-// GetStats gets connection, request, ssl, zone, and upstream related stats from the NGINX Plus API.
+// GetStats gets connection, request, ssl, zone, stream zone, upstream and stream upstream related stats from the NGINX Plus API.
 func (client *NginxClient) GetStats() (*Stats, error) {
 	cons, err := client.getConnections()
 	if err != nil {
@@ -594,17 +652,29 @@ func (client *NginxClient) GetStats() (*Stats, error) {
 		return nil, fmt.Errorf("failed to get stats: %v", err)
 	}
 
+	streamZones, err := client.getStreamServerZones()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get stats: %v", err)
+	}
+
 	upstreams, err := client.getUpstreams()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get stats: %v", err)
 	}
 
+	streamUpstreams, err := client.getStreamUpstreams()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get stats: %v", err)
+	}
+
 	return &Stats{
-		Connections:  *cons,
-		HTTPRequests: *requests,
-		SSL:          *ssl,
-		ServerZones:  *zones,
-		Upstreams:    *upstreams,
+		Connections:       *cons,
+		HTTPRequests:      *requests,
+		SSL:               *ssl,
+		ServerZones:       *zones,
+		StreamServerZones: *streamZones,
+		Upstreams:         *upstreams,
+		StreamUpstreams:   *streamUpstreams,
 	}, nil
 }
 
@@ -646,6 +716,15 @@ func (client *NginxClient) getServerZones() (*ServerZones, error) {
 	return &zones, err
 }
 
+func (client *NginxClient) getStreamServerZones() (*StreamServerZones, error) {
+	var zones StreamServerZones
+	err := client.get("stream/server_zones", &zones)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get stream server zones: %v", err)
+	}
+	return &zones, err
+}
+
 func (client *NginxClient) getUpstreams() (*Upstreams, error) {
 	var upstreams Upstreams
 	err := client.get("http/upstreams", &upstreams)
@@ -654,3 +733,12 @@ func (client *NginxClient) getUpstreams() (*Upstreams, error) {
 	}
 	return &upstreams, nil
 }
+
+func (client *NginxClient) getStreamUpstreams() (*StreamUpstreams, error) {
+	var upstreams StreamUpstreams
+	err := client.get("stream/upstreams", &upstreams)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get stream upstreams: %v", err)
+	}
+	return &upstreams, nil
+}

docker/nginx.conf

@@ -33,7 +33,13 @@ http {
 
 stream {
     upstream stream_test {
-        zone stream 64k;
+        zone stream_test 64k;
+    }
+
+    server {
+        listen 8081;
+        proxy_pass stream_test;
+        status_zone stream_test;
     }
 
 }

tests/client_test.go

@@ -453,6 +453,75 @@ func TestStats(t *testing.T) {
 	}
 }
 
+func TestStreamStats(t *testing.T) {
+	httpClient := &http.Client{}
+	c, err := client.NewNginxClient(httpClient, "http://127.0.0.1:8080/api")
+	if err != nil {
+		t.Fatalf("Error connecting to nginx: %v", err)
+	}
+
+	// need upstream for stats
+	server := client.StreamUpstreamServer{
+		Server: "127.0.0.1:8080",
+	}
+	err = c.AddStreamServer(streamUpstream, server)
+	if err != nil {
+		t.Errorf("Error adding stream upstream server: %v", err)
+	}
+
+	// make request so we have stream server zone stats, expect 5xx error
+	streamClient := &http.Client{}
+	_, err = streamClient.Get("http://127.0.0.1:8081")
+	if err != nil {
+		t.Errorf("Error making request: %v", err)
+	}
+
+	stats, err := c.GetStats()
+	if err != nil {
+		t.Errorf("Error getting stats: %v", err)
+	}
+
+	if stats.Connections.Active == 0 {
+		t.Errorf("Bad connections: %v", stats.Connections)
+	}
+	// SSL metrics blank in this example
+	if len(stats.StreamServerZones) < 1 {
+		t.Errorf("No StreamServerZone metrics: %v", stats.StreamServerZones)
+	}
+
+	if streamServerZone, ok := stats.StreamServerZones[streamUpstream]; ok {
+		if streamServerZone.Connections < 1 {
+			t.Errorf("StreamServerZone stats missing: %v", streamServerZone)
+		}
+	} else {
+		t.Errorf("StreamServerZone 'stream_test' not found")
+	}
+
+	if upstream, ok := stats.StreamUpstreams[streamUpstream]; ok {
+		if len(upstream.Peers) < 1 {
+			t.Errorf("stream upstream server not visible in stats")
+		} else {
+			if upstream.Peers[0].State != "up" {
+				t.Errorf("stream upstream server state should be 'up'")
+			}
+			if upstream.Peers[0].Connections < 0 {
+				t.Errorf("stream upstream should have connects value")
+			}
+			if upstream.Peers[0].HealthChecks.LastPassed {
+				t.Errorf("stream upstream server health check should report last failed")
+			}
+		}
+	} else {
+		t.Errorf("Stream upstream 'stream_test' not found")
+	}
+
+	// cleanup stream upstream servers
+	_, _, err = c.UpdateStreamServers(streamUpstream, []client.StreamUpstreamServer{})
+	if err != nil {
+		t.Errorf("Couldn't remove stream servers: %v", err)
+	}
+}
+
 func compareUpstreamServers(x []client.UpstreamServer, y []client.UpstreamServer) bool {
 	var xServers []string
 	for _, us := range x {