Add worker conn metrics

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.6.2
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/mwitkow/go-proto-validators v0.3.2
-	github.com/nginxinc/nginx-plus-go-client v0.10.0
+	github.com/nginxinc/nginx-plus-go-client v0.11.0
 	github.com/nginxinc/nginx-prometheus-exporter v0.11.0
 	github.com/nxadm/tail v1.4.8
 	github.com/orcaman/concurrent-map v1.0.0

go.sum

@@ -677,8 +677,8 @@ github.com/nbutton23/zxcvbn-go v0.0.0-20210217022336-fa2cb2858354 h1:4kuARK6Y6Fx
 github.com/nbutton23/zxcvbn-go v0.0.0-20210217022336-fa2cb2858354/go.mod h1:KSVJerMDfblTH7p5MZaTt+8zaT2iEk3AkVb9PQdZuE8=
 github.com/nginxinc/nginx-go-crossplane v0.4.24 h1:6gwZ8Boh8FS/kiZlfwSImO6UFnCjPeZ2uleHGSx4b+4=
 github.com/nginxinc/nginx-go-crossplane v0.4.24/go.mod h1:UzbZnyFv0vPlt1Urbnp/mrFCzBL4tYCReFuNBpFQEfI=
-github.com/nginxinc/nginx-plus-go-client v0.10.0 h1:3zsMMkPvRDo8D7ZSprXtbAEW/SDmezZWzxdyS+6oAlc=
-github.com/nginxinc/nginx-plus-go-client v0.10.0/go.mod h1:0v3RsQCvRn/IyrMtW+DK6CNkz+PxEsXDJPjQ3yUMBF0=
+github.com/nginxinc/nginx-plus-go-client v0.11.0 h1:XxKag3dlcF1gA0HgGsrJktbzBW4W+s369PIgT3lSR2c=
+github.com/nginxinc/nginx-plus-go-client v0.11.0/go.mod h1:UvrcgWbUWEJzvbstNnPfq8Ogz9BTi1gHzlQ2ebAJisM=
 github.com/nginxinc/nginx-prometheus-exporter v0.11.0 h1:21xjnqNgxtni2jDgAQ90bl15uDnrTreO9sIlu1YsX/U=
 github.com/nginxinc/nginx-prometheus-exporter v0.11.0/go.mod h1:GdyHnWAb8q8OW1Pssrrqbcqra0SH0Vn6UXICMmyWkw8=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=

src/core/metrics/collectors/nginx_test.go

@@ -12,16 +12,15 @@ import (
 	"reflect"
 	"sync"
 	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
 
 	"github.com/nginx/agent/sdk/v2/proto"
 	"github.com/nginx/agent/v2/src/core"
 	"github.com/nginx/agent/v2/src/core/config"
 	"github.com/nginx/agent/v2/src/core/metrics"
 	tutils "github.com/nginx/agent/v2/test/utils"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 )
 
 var (
@@ -49,10 +48,13 @@ var (
 		},
 		Features: config.Defaults.Features,
 		Nginx: config.Nginx{
-			Debug:               false,
-			NginxCountingSocket: "unix:/var/run/nginx-agent/nginx.sock",
+			Debug:                 false,
+			NginxCountingSocket:   "unix:/var/run/nginx-agent/nginx.sock",
+			NginxClientVersion:    9,
+			TreatWarningsAsErrors: false,
 		},
 	}
+
 	collectorConfigNoApi = &metrics.NginxCollectorConfig{
 		BinPath:            "/path/to/nginx",
 		NginxId:            nginxId,
@@ -174,8 +176,8 @@ func TestNginxCollector_Collect(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go nginxCollector.Collect(ctx, wg, make(chan<- *metrics.StatsEntityWrapper))
+	wg.Wait()
 
-	time.Sleep(10 * time.Millisecond)
 	mockNginxSource1.AssertExpectations(t)
 	mockNginxSource2.AssertExpectations(t)
 }

src/core/metrics/sources/nginx_plus.go

@@ -9,10 +9,13 @@ package sources
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"io"
 	"math"
 	"net/http"
 	"sync"
+	"time"
 
 	"github.com/nginx/agent/sdk/v2/proto"
 	"github.com/nginx/agent/v2/src/core/metrics"
@@ -29,8 +32,11 @@ const (
 	peerStateUnavail   = "unavail"
 	peerStateChecking  = "checking"
 	peerStateUnhealthy = "unhealthy"
-	valueFloat64One    = float64(1)
-	valueFloat64Zero   = float64(0)
+
+	valueFloat64One  = float64(1)
+	valueFloat64Zero = float64(0)
+
+	minimumAPIVersion = 7 // NGINX Plus R25+
 )
 
 // NginxPlus generates metrics from NGINX Plus API
@@ -53,31 +59,39 @@ func NewNginxPlus(baseDimensions *metrics.CommonDim, nginxNamespace, plusNamespa
 func (c *NginxPlus) Collect(ctx context.Context, wg *sync.WaitGroup, m chan<- *metrics.StatsEntityWrapper) {
 	defer wg.Done()
 	c.init.Do(func() {
-		cl, err := plusclient.NewNginxClientWithVersion(&http.Client{}, c.plusAPI, c.clientVersion)
+		latestAPIVersion, err := getLatestAPIVersion(ctx, &http.Client{}, c.plusAPI)
+		if err != nil {
+			c.logger.Log(fmt.Sprintf("Failed to check available api versions: %v", err))
+			SendNginxDownStatus(ctx, c.baseDimensions.ToDimensions(), m)
+			return
+		}
+
+		cl, err := plusclient.NewNginxClientWithVersion(&http.Client{}, c.plusAPI, latestAPIVersion)
 		if err != nil {
-			c.logger.Log(fmt.Sprintf("Failed to create plus metrics client, %v", err))
+			c.logger.Log(fmt.Sprintf("Failed to create plus metrics client: %v", err))
 			SendNginxDownStatus(ctx, c.baseDimensions.ToDimensions(), m)
 			return
 		}
+
 		c.prevStats, err = cl.GetStats()
 		if err != nil {
-			c.logger.Log(fmt.Sprintf("Failed to retrieve plus metrics, %v", err))
+			c.logger.Log(fmt.Sprintf("Failed to retrieve plus metrics: %v", err))
 			SendNginxDownStatus(ctx, c.baseDimensions.ToDimensions(), m)
 			c.prevStats = nil
 			return
 		}
 	})
 
-	cl, err := plusclient.NewNginxClientWithVersion(&http.Client{}, c.plusAPI, c.clientVersion)
+	cl, err := plusclient.NewNginxClient(&http.Client{}, c.plusAPI)
 	if err != nil {
-		c.logger.Log(fmt.Sprintf("Failed to create plus metrics client, %v", err))
+		c.logger.Log(fmt.Sprintf("Failed to create plus metrics client: %v", err))
 		SendNginxDownStatus(ctx, c.baseDimensions.ToDimensions(), m)
 		return
 	}
 
 	stats, err := cl.GetStats()
 	if err != nil {
-		c.logger.Log(fmt.Sprintf("Failed to retrieve plus metrics, %v", err))
+		c.logger.Log(fmt.Sprintf("Failed to retrieve plus metrics: %v", err))
 		SendNginxDownStatus(ctx, c.baseDimensions.ToDimensions(), m)
 		return
 	}
@@ -130,6 +144,7 @@ func (c *NginxPlus) collectMetrics(stats, prevStats *plusclient.Stats) (entries
 	entries = append(entries, c.cacheMetrics(stats, prevStats)...)
 	entries = append(entries, c.httpUpstreamMetrics(stats, prevStats)...)
 	entries = append(entries, c.streamUpstreamMetrics(stats, prevStats)...)
+	entries = append(entries, c.workerMetrics(stats, prevStats)...)
 
 	return
 }
@@ -801,6 +816,30 @@ func (c *NginxPlus) httpLimitRequestMetrics(stats, prevStats *plusclient.Stats)
 	return limitRequestMetrics
 }
 
+func (c *NginxPlus) workerMetrics(stats, prevStats *plusclient.Stats) []*metrics.StatsEntityWrapper {
+	workerMetrics := make([]*metrics.StatsEntityWrapper, 0)
+
+	for _, w := range stats.Workers {
+		l := &namedMetric{namespace: c.plusNamespace, group: "worker"}
+
+		simpleMetrics := l.convertSamplesToSimpleMetrics(map[string]float64{
+			"id":                   float64(w.ID),
+			"process_id":           float64(w.ProcessID),
+			"conn.accepted":        float64(w.Connections.Accepted),
+			"conn.dropped":         float64(w.Connections.Dropped),
+			"conn.active":          float64(w.Connections.Active),
+			"conn.idle":            float64(w.Connections.Idle),
+			"http.request.total":   float64(w.HTTP.HTTPRequests.Total),
+			"http.request.current": float64(w.HTTP.HTTPRequests.Current),
+		})
+
+		dims := c.baseDimensions.ToDimensions()
+		workerMetrics = append(workerMetrics, metrics.NewStatsEntityWrapper(dims, simpleMetrics, proto.MetricsReport_INSTANCE))
+	}
+
+	return workerMetrics
+}
+
 func getHttpUpstreamPeerKey(peer plusclient.Peer) (key string) {
 	key = fmt.Sprintf("%s-%s-%s", peer.Server, peer.Service, peer.Name)
 	return
@@ -827,10 +866,48 @@ func createStreamPeerMap(peers []plusclient.StreamPeer) map[string]plusclient.St
 	return m
 }
 
-func boolToFloat64(mybool bool) float64 {
-	if mybool {
+func boolToFloat64(myBool bool) float64 {
+	if myBool {
 		return valueFloat64One
 	} else {
 		return valueFloat64Zero
 	}
 }
+
+func getLatestAPIVersion(ctx context.Context, httpClient *http.Client, endpoint string) (int, error) {
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+	if err != nil {
+		return 0, fmt.Errorf("failed to create a get request: %w", err)
+	}
+
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		return 0, fmt.Errorf("%v is not accessible: %w", endpoint, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("%v is not accessible: expected %v response, got %v", endpoint, http.StatusOK, resp.StatusCode)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return 0, fmt.Errorf("error while reading body of the response: %w", err)
+	}
+
+	var vers []int
+	err = json.Unmarshal(body, &vers)
+	if err != nil {
+		return 0, fmt.Errorf("error unmarshalling versions, got %q response: %w", string(body), err)
+	}
+
+	latestAPIVer := vers[len(vers)-1]
+	if latestAPIVer < minimumAPIVersion {
+		return 0, fmt.Errorf("%v is an unsupported api version. Must be at least version %v", endpoint, minimumAPIVersion)
+	}
+
+	return latestAPIVer, nil
+}

src/core/metrics/sources/nginx_plus_test.go

@@ -331,6 +331,24 @@ func (f *FakeNginxPlus) Collect(ctx context.Context, wg *sync.WaitGroup, m chan<
 				RejectedDryRun: 2,
 			},
 		},
+		Workers: []*plusclient.Workers{
+			{
+				ID:        21,
+				ProcessID: 12345,
+				HTTP: plusclient.WorkersHTTP{
+					HTTPRequests: plusclient.HTTPRequests{
+						Total:   112,
+						Current: 213,
+					},
+				},
+				Connections: plusclient.Connections{
+					Accepted: 21,
+					Dropped:  25,
+					Active:   12,
+					Idle:     1,
+				},
+			},
+		},
 	}
 
 	prevStats := plusclient.Stats{
@@ -430,6 +448,24 @@ func (f *FakeNginxPlus) Collect(ctx context.Context, wg *sync.WaitGroup, m chan<
 				RejectedDryRun: 0,
 			},
 		},
+		Workers: []*plusclient.Workers{
+			{
+				ID:        0,
+				ProcessID: 0,
+				HTTP: plusclient.WorkersHTTP{
+					HTTPRequests: plusclient.HTTPRequests{
+						Total:   0,
+						Current: 0,
+					},
+				},
+				Connections: plusclient.Connections{
+					Accepted: 0,
+					Dropped:  0,
+					Active:   0,
+					Idle:     0,
+				},
+			},
+		},
 	}
 
 	f.baseDimensions.NginxType = "plus"
@@ -701,16 +737,30 @@ func TestNginxPlus_Collect(t *testing.T) {
 		"plus.http.limit_reqs.rejected_dry_run": 2,
 	}
 
-	hostInfo := &proto.HostInfo{
-		Hostname: "MyServer",
+	expectedWorkerMetrics := map[string]float64{
+		"plus.worker.id":                   21,
+		"plus.worker.process_id":           12345,
+		"plus.worker.conn.accepted":        21,
+		"plus.worker.conn.dropped":         25,
+		"plus.worker.conn.active":          12,
+		"plus.worker.conn.idle":            1,
+		"plus.worker.http.request.total":   112,
+		"plus.worker.http.request.current": 213,
 	}
+
 	tests := []struct {
 		baseDimensions *metrics.CommonDim
 		m              chan *metrics.StatsEntityWrapper
 	}{
 		{
-			baseDimensions: metrics.NewCommonDim(hostInfo, &config.Config{}, ""),
-			m:              make(chan *metrics.StatsEntityWrapper, 127),
+			baseDimensions: metrics.NewCommonDim(
+				&proto.HostInfo{
+					Hostname: "MyServer",
+				},
+				&config.Config{},
+				"",
+			),
+			m: make(chan *metrics.StatsEntityWrapper, 127),
 		},
 	}
 
@@ -1148,6 +1198,12 @@ func TestNginxPlus_Collect(t *testing.T) {
 			}
 		}
 
+		workerMetrics := <-test.m
+		for _, metric := range workerMetrics.Data.Simplemetrics {
+			assert.Contains(t, expectedWorkerMetrics, metric.Name)
+			assert.Equal(t, expectedWorkerMetrics[metric.Name], metric.Value)
+		}
+
 		var extraMetrics []*metrics.StatsEntityWrapper
 	EMWAIT:
 		for {

test/performance/go.mod

@@ -44,7 +44,7 @@ require (
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nginxinc/nginx-go-crossplane v0.4.24 // indirect
-	github.com/nginxinc/nginx-plus-go-client v0.10.0 // indirect
+	github.com/nginxinc/nginx-plus-go-client v0.11.0 // indirect
 	github.com/nginxinc/nginx-prometheus-exporter v0.11.0 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/orcaman/concurrent-map v1.0.0 // indirect

test/performance/go.sum

@@ -199,8 +199,8 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nginxinc/nginx-go-crossplane v0.4.24 h1:6gwZ8Boh8FS/kiZlfwSImO6UFnCjPeZ2uleHGSx4b+4=
 github.com/nginxinc/nginx-go-crossplane v0.4.24/go.mod h1:UzbZnyFv0vPlt1Urbnp/mrFCzBL4tYCReFuNBpFQEfI=
-github.com/nginxinc/nginx-plus-go-client v0.10.0 h1:3zsMMkPvRDo8D7ZSprXtbAEW/SDmezZWzxdyS+6oAlc=
-github.com/nginxinc/nginx-plus-go-client v0.10.0/go.mod h1:0v3RsQCvRn/IyrMtW+DK6CNkz+PxEsXDJPjQ3yUMBF0=
+github.com/nginxinc/nginx-plus-go-client v0.11.0 h1:XxKag3dlcF1gA0HgGsrJktbzBW4W+s369PIgT3lSR2c=
+github.com/nginxinc/nginx-plus-go-client v0.11.0/go.mod h1:UvrcgWbUWEJzvbstNnPfq8Ogz9BTi1gHzlQ2ebAJisM=
 github.com/nginxinc/nginx-prometheus-exporter v0.11.0 h1:21xjnqNgxtni2jDgAQ90bl15uDnrTreO9sIlu1YsX/U=
 github.com/nginxinc/nginx-prometheus-exporter v0.11.0/go.mod h1:GdyHnWAb8q8OW1Pssrrqbcqra0SH0Vn6UXICMmyWkw8=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=