Add stream upstream, stream server zones metrics support

* Add stream upstream/server zone metrics support
* Extend GetStats to include stream metrics
* Added integration test to validate that stream metrics are returned
* Handle missing stream block
* Add sleep to wait for health checks to pass
* Simplify internal error handling
* Add internalError.Wrap for preserving context

Author: Dean-Coakley

Makefile

@@ -1,17 +1,24 @@
 NGINX_PLUS_VERSION=15-2
 NGINX_IMAGE=nginxplus:$(NGINX_PLUS_VERSION)
 
-test: docker-build run-nginx-plus test-run clean
-	
+test: docker-build run-nginx-plus test-run configure-no-stream-block test-run-no-stream-block clean
+
 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/*
-	go test tests/client_test.go
+	GOCACHE=off go test tests/client_test.go
+
+configure-no-stream-block:
+	docker cp docker/nginx_no_stream.conf nginx-plus-test:/etc/nginx/nginx.conf
+	docker exec nginx-plus-test nginx -s reload
+
+test-run-no-stream-block:
+	GOCACHE=off go test tests/client_no_stream_test.go
 
 clean:
 	docker kill nginx-plus-test

client/nginx.go

@@ -12,6 +12,8 @@ import (
 // APIVersion is a version of NGINX Plus API.
 const APIVersion = 2
 
+const streamNotConfiguredCode = "StreamNotConfigured"
+
 // NginxClient lets you access NGINX Plus API.
 type NginxClient struct {
 	apiEndpoint string
@@ -57,14 +59,33 @@ type apiError struct {
 	Code   string
 }
 
+type internalError struct {
+	apiError
+	err string
+}
+
+// Error allows internalError to match the Error interface.
+func (internalError *internalError) Error() string {
+	return internalError.err
+}
+
+// Wrap is a way of including current context while preserving previous error information,
+// similar to `return fmt.Errof("error doing foo, err: %v", err)` but for our internalError type.
+func (internalError *internalError) Wrap(err string) *internalError {
+	internalError.err = fmt.Sprintf("%v. %v", err, internalError.err)
+	return internalError
+}
+
 // 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 +122,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 +145,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 +165,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 +207,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
@@ -214,13 +291,18 @@ func getAPIVersions(httpClient *http.Client, endpoint string) (*versions, error)
 	return &vers, nil
 }
 
-func createResponseMismatchError(respBody io.ReadCloser, mainErr error) error {
-	apiErr, err := readAPIErrorResponse(respBody)
+func createResponseMismatchError(respBody io.ReadCloser) *internalError {
+	apiErrResp, err := readAPIErrorResponse(respBody)
 	if err != nil {
-		return fmt.Errorf("%v; failed to read the response body: %v", mainErr, err)
+		return &internalError{
+			err: fmt.Sprintf("failed to read the response body: %v", err),
+		}
 	}
 
-	return fmt.Errorf("%v; error: %v", mainErr, apiErr.toString())
+	return &internalError{
+		err:      apiErrResp.toString(),
+		apiError: apiErrResp.Error,
+	}
 }
 
 func readAPIErrorResponse(respBody io.ReadCloser) (*apiErrorResponse, error) {
@@ -379,8 +461,9 @@ func (client *NginxClient) get(path string, data interface{}) error {
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != http.StatusOK {
-		mainErr := fmt.Errorf("expected %v response, got %v", http.StatusOK, resp.StatusCode)
-		return createResponseMismatchError(resp.Body, mainErr)
+		return createResponseMismatchError(resp.Body).Wrap(fmt.Sprintf(
+			"expected %v response, got %v",
+			http.StatusOK, resp.StatusCode))
 	}
 
 	body, err := ioutil.ReadAll(resp.Body)
@@ -409,8 +492,9 @@ func (client *NginxClient) post(path string, input interface{}) error {
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != http.StatusCreated {
-		mainErr := fmt.Errorf("expected %v response, got %v", http.StatusCreated, resp.StatusCode)
-		return createResponseMismatchError(resp.Body, mainErr)
+		return createResponseMismatchError(resp.Body).Wrap(fmt.Sprintf(
+			"expected %v response, got %v",
+			http.StatusCreated, resp.StatusCode))
 	}
 
 	return nil
@@ -431,9 +515,9 @@ func (client *NginxClient) delete(path string) error {
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		mainErr := fmt.Errorf("failed to complete delete request: expected %v response, got %v",
-			http.StatusOK, resp.StatusCode)
-		return createResponseMismatchError(resp.Body, mainErr)
+		return createResponseMismatchError(resp.Body).Wrap(fmt.Sprintf(
+			"failed to complete delete request: expected %v response, got %v",
+			http.StatusOK, resp.StatusCode))
 	}
 	return nil
 }
@@ -458,7 +542,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 +656,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 {
@@ -599,12 +683,24 @@ 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)
+	}
+
+	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 +742,20 @@ 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 {
+		if err, ok := err.(*internalError); ok {
+			if err.Code == streamNotConfiguredCode {
+				return &zones, 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 +764,17 @@ 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 {
+		if err, ok := err.(*internalError); ok {
+			if err.Code == streamNotConfiguredCode {
+				return &upstreams, nil
+			}
+		}
+		return nil, fmt.Errorf("failed to get stream upstreams: %v", err)
+	}
+	return &upstreams, nil
+}

docker/nginx.conf

@@ -33,7 +33,14 @@ http {
 
 stream {
     upstream stream_test {
-        zone stream 64k;
+        zone stream_test 64k;
+    }
+
+    server {
+        listen 8081;
+        proxy_pass stream_test;
+        status_zone stream_test;
+        health_check interval=10 fails=3 passes=1;
     }
 
 }

docker/nginx_no_stream.conf

@@ -0,0 +1,32 @@
+
+user  nginx;
+worker_processes  auto;
+
+error_log  /var/log/nginx/error.log notice;
+pid        /var/run/nginx.pid;
+
+
+events {
+    worker_connections  1024;
+}
+
+
+http {
+    include       /etc/nginx/mime.types;
+    default_type  application/octet-stream;
+
+    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
+                      '$status $body_bytes_sent "$http_referer" '
+                      '"$http_user_agent" "$http_x_forwarded_for"';
+
+    access_log  /var/log/nginx/access.log  main;
+
+    sendfile        on;
+    #tcp_nopush     on;
+
+    keepalive_timeout  65;
+
+    #gzip  on;
+
+    include /etc/nginx/conf.d/*.conf;
+}

docker/test.conf

@@ -22,4 +22,4 @@ server {
         health_check interval=10 fails=3 passes=1;
     }
     status_zone test;
-}
+}