Add zone sync endpoint to stats

Makefile

@@ -10,7 +10,9 @@ 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 -p 8081:8081 $(NGINX_IMAGE)
+	docker network create --driver bridge test
+	docker run --network=test -d --name nginx-plus-test --network-alias=nginx-plus-test --rm -p 8080:8080 -p 8081:8081 $(NGINX_IMAGE)
+	docker run --network=test -d --name nginx-plus-test-helper --network-alias=nginx-plus-test --rm -p 8090:8080 -p 8091:8081 $(NGINX_IMAGE)
 
 test-run:
 	go test client/*
@@ -26,4 +28,6 @@ test-run-no-stream-block:
 	go test tests/client_no_stream_test.go
 
 clean:
-	docker kill nginx-plus-test
+	-docker kill nginx-plus-test
+	-docker kill nginx-plus-test-helper
+	-docker network rm test

README.md

@@ -40,7 +40,7 @@ Run Tests:
 $ make test
 ```
 
-This will build and run an NGINX Plus container, execute the client tests against NGINX Plus API, and then clean up. If it fails and you want to clean up (i.e. stop the running container), please use `$ make clean`
+This will build and run two NGINX Plus containers and create one docker network of type bridge, execute the client tests against both NGINX Plus APIs, and then clean up. If it fails and you want to clean up (i.e. stop the running containers and remove the docker network), please use `$ make clean`
 
 ## Support
 This project is not covered by the NGINX Plus support contract.

client/nginx.go

@@ -88,6 +88,7 @@ type Stats struct {
 	Upstreams         Upstreams
 	StreamServerZones StreamServerZones
 	StreamUpstreams   StreamUpstreams
+	StreamZoneSync    StreamZoneSync
 }
 
 // NginxInfo contains general information about NGINX Plus.
@@ -149,6 +150,27 @@ type StreamServerZone struct {
 	Sent        uint64
 }
 
+// StreamZoneSync represents the sync information per each shared memory zone and the sync information per node in a cluster
+type StreamZoneSync struct {
+	Zones  map[string]SyncZone
+	Status StreamZoneSyncStatus
+}
+
+// SyncZone represents the syncronization status of a shared memory zone
+type SyncZone struct {
+	RecordsPending uint64 `json:"records_pending"`
+	RecordsTotal   uint64 `json:"records_total"`
+}
+
+// StreamZoneSyncStatus represents the status of a shared memory zone
+type StreamZoneSyncStatus struct {
+	BytesIn     uint64 `json:"bytes_in"`
+	MsgsIn      uint64 `json:"msgs_in"`
+	MsgsOut     uint64 `json:"msgs_out"`
+	BytesOut    uint64 `json:"bytes_out"`
+	NodesOnline uint64 `json:"nodes_online"`
+}
+
 // Responses represents HTTP response related stats.
 type Responses struct {
 	Responses1xx uint64 `json:"1xx"`
@@ -728,6 +750,11 @@ func (client *NginxClient) GetStats() (*Stats, error) {
 		return nil, fmt.Errorf("failed to get stats: %v", err)
 	}
 
+	streamZoneSync, err := client.getStreamZoneSync()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get stats: %v", err)
+	}
+
 	return &Stats{
 		NginxInfo:         *info,
 		Connections:       *cons,
@@ -737,6 +764,7 @@ func (client *NginxClient) GetStats() (*Stats, error) {
 		StreamServerZones: *streamZones,
 		Upstreams:         *upstreams,
 		StreamUpstreams:   *streamUpstreams,
+		StreamZoneSync:    *streamZoneSync,
 	}, nil
 }
 
@@ -822,6 +850,22 @@ func (client *NginxClient) getStreamUpstreams() (*StreamUpstreams, error) {
 	return &upstreams, nil
 }
 
+func (client *NginxClient) getStreamZoneSync() (*StreamZoneSync, error) {
+	var streamZoneSync StreamZoneSync
+	err := client.get("stream/zone_sync", &streamZoneSync)
+	if err != nil {
+		if err, ok := err.(*internalError); ok {
+
+			if err.Code == pathNotFoundCode {
+				return &streamZoneSync, nil
+			}
+		}
+		return nil, fmt.Errorf("failed to get stream zone sync: %v", err)
+	}
+
+	return &streamZoneSync, err
+}
+
 // KeyValPairs are the key-value pairs stored in a zone.
 type KeyValPairs map[string]string
 

docker/nginx.conf

@@ -37,6 +37,7 @@ http {
 stream {
     keyval_zone zone=zone_one_stream:32k;
     keyval $hostname $text zone=zone_one_stream;
+    keyval_zone zone=zone_test_sync:32k timeout=5s sync;
 
     upstream stream_test {
         zone stream_test 64k;
@@ -49,4 +50,12 @@ stream {
         health_check interval=10 fails=3 passes=1;
     }
 
+    resolver 127.0.0.11 valid=5s;
+
+    server {
+        listen 7777;
+
+        zone_sync;
+        zone_sync_server nginx-plus-test:7777 resolve;
+    }
 }

tests/client_test.go

@@ -13,6 +13,7 @@ import (
 const (
 	upstream       = "test"
 	streamUpstream = "stream_test"
+	streamZoneSync = "zone_test_sync"
 )
 
 func TestStreamClient(t *testing.T) {
@@ -673,7 +674,8 @@ func TestKeyValueStream(t *testing.T) {
 		t.Errorf("Couldn't get keyvals, %v", err)
 	}
 	expectedKeyValuePairsByZone := client.KeyValPairsByZone{
-		zoneName: expectedKeyValPairs,
+		zoneName:       expectedKeyValPairs,
+		streamZoneSync: client.KeyValPairs{},
 	}
 	if !reflect.DeepEqual(expectedKeyValuePairsByZone, keyValPairsByZone) {
 		t.Errorf("maps are not equal. expected: %+v, got: %+v", expectedKeyValuePairsByZone, keyValPairsByZone)
@@ -741,6 +743,98 @@ func TestKeyValueStream(t *testing.T) {
 	}
 }
 
+func TestStreamZoneSync(t *testing.T) {
+	c1, err := client.NewNginxClient(&http.Client{}, "http://127.0.0.1:8080/api")
+	if err != nil {
+		t.Fatalf("Error connecting to nginx: %v", err)
+	}
+
+	c2, err := client.NewNginxClient(&http.Client{}, "http://127.0.0.1:8090/api")
+	if err != nil {
+		t.Fatalf("Error connecting to nginx: %v", err)
+	}
+
+	err = c1.AddStreamKeyValPair(streamZoneSync, "key1", "val1")
+	if err != nil {
+		t.Errorf("Couldn't set keyvals: %v", err)
+	}
+
+	// wait for nodes to sync information of synced zones
+	time.Sleep(5 * time.Second)
+
+	statsC1, err := c1.GetStats()
+	if err != nil {
+		t.Errorf("Error getting stats: %v", err)
+	}
+
+	if statsC1.StreamZoneSync.Status.NodesOnline == 0 {
+		t.Errorf("At least 1 node must be online")
+	}
+
+	if statsC1.StreamZoneSync.Status.MsgsOut == 0 {
+		t.Errorf("Msgs out cannot be 0")
+	}
+
+	if statsC1.StreamZoneSync.Status.MsgsIn == 0 {
+		t.Errorf("Msgs in cannot be 0")
+	}
+
+	if statsC1.StreamZoneSync.Status.BytesIn == 0 {
+		t.Errorf("Bytes in cannot be 0")
+	}
+
+	if statsC1.StreamZoneSync.Status.BytesOut == 0 {
+		t.Errorf("Bytes Out cannot be 0")
+	}
+
+	if zone, ok := statsC1.StreamZoneSync.Zones[streamZoneSync]; ok {
+		if zone.RecordsTotal == 0 {
+			t.Errorf("Total records cannot be 0 after adding keyvals")
+		}
+		if zone.RecordsPending != 0 {
+			t.Errorf("Pending records must be 0 after adding keyvals")
+		}
+	} else {
+		t.Errorf("Sync zone %v missing in stats", streamZoneSync)
+	}
+
+	statsC2, err := c2.GetStats()
+	if err != nil {
+		t.Errorf("Error getting stats: %v", err)
+	}
+
+	if statsC2.StreamZoneSync.Status.NodesOnline == 0 {
+		t.Errorf("At least 1 node must be online")
+	}
+
+	if statsC2.StreamZoneSync.Status.MsgsOut != 0 {
+		t.Errorf("Msgs out must be 0")
+	}
+
+	if statsC2.StreamZoneSync.Status.MsgsIn == 0 {
+		t.Errorf("Msgs in cannot be 0")
+	}
+
+	if statsC2.StreamZoneSync.Status.BytesIn == 0 {
+		t.Errorf("Bytes in cannot be 0")
+	}
+
+	if statsC2.StreamZoneSync.Status.BytesOut != 0 {
+		t.Errorf("Bytes out must be 0")
+	}
+
+	if zone, ok := statsC2.StreamZoneSync.Zones[streamZoneSync]; ok {
+		if zone.RecordsTotal == 0 {
+			t.Errorf("Total records cannot be 0 after adding keyvals")
+		}
+		if zone.RecordsPending != 0 {
+			t.Errorf("Pending records must be 0 after adding keyvals")
+		}
+	} else {
+		t.Errorf("Sync zone %v missing in stats", streamZoneSync)
+	}
+}
+
 func compareUpstreamServers(x []client.UpstreamServer, y []client.UpstreamServer) bool {
 	var xServers []string
 	for _, us := range x {