APIGOV-26694 - send in metric detail for each received transaction (#41)

* updates for using metrics

* update func name

* updates for tests

* updates for test coverage

* update deps

* add non-verbose target for test execution

* add wait group to tests

* use new sdk changes for sending events on shutdown

* update deps

* use local interface for metric collector

* updates for helm charts

* make consumer tags an interface

* use statefulset for ta

* update

* logic fix

* fixes from teting

* remove old files

* remove old files

* remove unused vars from yaml

* add s3 sync docs

* update stage

* fix consumer metrics

* more readme updates

* fix quota tests

* update dep

* update appVersion

* add note and typos

.gitignore

@@ -14,7 +14,7 @@ bin/
 .idea/
 **.tgz
 .log
-.env
+.env*
 *.pem
 *.key
 .DS_Store

Makefile

@@ -19,6 +19,10 @@ all: clean
 test: dep
 	@go vet ${GO_PKG_LIST}
 	@go test -race -v -short -coverprofile=${WORKSPACE}/gocoverage.out -count=1 ${GO_PKG_LIST}
+
+test-s: dep
+	@go vet ${GO_PKG_LIST}
+	@go test -race -short -coverprofile=${WORKSPACE}/gocoverage.out -count=1 ${GO_PKG_LIST}
 
 clean:
 	@rm -rf ./bin/

build/traceability/kong_traceability_agent.yml

@@ -1,15 +1,10 @@
 kong_traceability_agent:
   # Settings for connecting to Kong
   kong:
-    admin:
-      url: ${KONG_ADMIN_URL}
-      auth:
-        apikey:
-          header: ${KONG_ADMIN_AUTH_APIKEY_HEADER}
-          value: ${KONG_ADMIN_AUTH_APIKEY_VALUE}
-    httpLogPlugin:
-      path: ${KONG_HTTPLOGPLUGIN_PATH}
-      port: ${KONG_HTTPLOGPLUGIN_PORT}
+    logs:
+      http:
+        path: ${KONG_LOGS_HTTP_PATH}
+        port: ${KONG_LOGS_HTTP_PORT}
   # Settings for connecting to Amplify Central
   central:
     url: ${CENTRAL_URL:https://apicentral.axway.com}

go.mod

@@ -3,7 +3,7 @@ module github.com/Axway/agents-kong
 go 1.18
 
 require (
-	github.com/Axway/agent-sdk v1.1.68-0.20231120204404-5fa2c5b231e1
+	github.com/Axway/agent-sdk v1.1.69
 	github.com/elastic/beats/v7 v7.17.15
 	github.com/google/uuid v1.3.1
 	github.com/kong/go-kong v0.47.0

go.sum

@@ -36,8 +36,8 @@ cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RX
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
 cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3fOKtUw0Xmo=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
-github.com/Axway/agent-sdk v1.1.68-0.20231120204404-5fa2c5b231e1 h1:UPRwbD7Hb/p4CpSHv0hDwOmjbSqYy5Fk7pSnnatDwes=
-github.com/Axway/agent-sdk v1.1.68-0.20231120204404-5fa2c5b231e1/go.mod h1:Iuv9KlWksVTbTKdfs4bKVYMDc33ZTLYoHt572z2CbbI=
+github.com/Axway/agent-sdk v1.1.69 h1:k9YHhoXfGoZtBw6hJf3TrRNPdmLTY07eXewY5Odxx+M=
+github.com/Axway/agent-sdk v1.1.69/go.mod h1:Iuv9KlWksVTbTKdfs4bKVYMDc33ZTLYoHt572z2CbbI=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
 github.com/Azure/go-autorest/autorest v0.11.12/go.mod h1:eipySxLmqSyC5s5k1CLupqet0PSENBEDP93LQ9a8QYw=

helm/kong-agents/Chart.yaml

@@ -19,10 +19,10 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.0.0
+version: 0.0.1
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: "0.0.0"
+appVersion: "v0.0.1"

helm/kong-agents/templates/_helpers.tpl

@@ -42,6 +42,14 @@ app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
 app.kubernetes.io/managed-by: {{ .Release.Service }}
 {{- end }}
 
+{{/*
+Traceability selector labels
+*/}}
+{{- define "kong-agents.traceability.selectorLabels" -}}
+{{ include "kong-agents.selectorLabels" . }}
+app.agent.type: traceability
+{{- end }}
+
 {{/*
 Selector labels
 */}}

helm/kong-agents/templates/discovery-deployment.yaml

@@ -121,7 +121,10 @@ spec:
                                                      "KONG_PROXY_PORTS_HTTP" 
                                                      "KONG_PROXY_PORTS_HTTPS" 
                                                      "KONG_SPEC_LOCALPATH"
-                                                     "KONG_SPEC_URLPATHS")))
+                                                     "KONG_SPEC_URLPATHS"
+                                                     "KONG_LOGS_HTTP_SERVER_PATH"
+                                                     "KONG_LOGS_HTTP_SERVER_PORT"
+                                                     "STATUS_PORT")))
            }} 
             - name: {{ $key }}
               value: {{ $value | quote }} 
@@ -132,6 +135,8 @@ spec:
               value: "/keys/private_key.pem"
             - name: CENTRAL_AUTH_PUBLICKEY
               value: "/keys/public_key.pem"
+            - name: STATUS_PORT
+              value: "{{ .Values.statusPort }}"
           volumeMounts:
             - name: "kong-agent-keys"
               mountPath: "/keys"

helm/kong-agents/templates/service.yaml

@@ -0,0 +1,17 @@
+{{- if .Values.kong.enable.traceability }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "kong-agents.fullname" . }}-traceability
+  labels:
+    {{- include "kong-agents.labels" . | nindent 4 }}
+spec:
+  selector:
+    {{- include "kong-agents.selectorLabels" . | nindent 4 }}
+    app.agent.type: traceability
+  ports:
+  - name: logs
+    protocol: TCP
+    port: {{ .Values.kong.logs.http.port }}
+    targetPort: logs
+{{- end -}}

helm/kong-agents/templates/traceability-deployment.yaml → helm/kong-agents/templates/traceability-statefulset.yaml

@@ -1,6 +1,6 @@
 {{- if .Values.kong.enable.traceability }}
 apiVersion: apps/v1
-kind: Deployment
+kind: StatefulSet
 metadata:
   name: {{ include "kong-agents.fullname" . }}-traceability
   labels:
@@ -9,7 +9,7 @@ spec:
   replicas: {{ .Values.traceability.replicaCount }}
   selector:
     matchLabels:
-      {{- include "kong-agents.selectorLabels" . | nindent 6 }}
+      {{- include "kong-agents.traceability.selectorLabels" . | nindent 6 }}
       {{- with .Values.additionalLabels }}
       {{- range $key, $value := . }}
       {{ default "none" $key }}: {{ default "none" $value | quote }}
@@ -22,7 +22,7 @@ spec:
         {{- toYaml . | nindent 8 }}
       {{- end }}
       labels:
-        {{- include "kong-agents.selectorLabels" . | nindent 8 }}
+        {{- include "kong-agents.traceability.selectorLabels" . | nindent 8 }}
         {{- with .Values.additionalLabels }}
         {{- range $key, $value := . }}
         {{ default "none" $key }}: {{ default "none" $value | quote }}
@@ -48,6 +48,9 @@ spec:
           - name: probe-port
             containerPort: {{ .Values.statusPort }}
             protocol: TCP
+          - name: logs
+            containerPort: {{ .Values.kong.logs.http.port }}
+            protocol: TCP
           livenessProbe:
             httpGet:
               path: /status
@@ -67,20 +70,10 @@ spec:
             {{- end }}
             {{- end }}
           env:
-            - name: KONG_ADMIN_URL
-              value: "{{ .Values.kong.admin.url }}"
-            {{- if .Values.kong.admin.auth.apikey.value }}
-            - name: KONG_ADMIN_AUTH_APIKEY_VALUE
-              valueFrom:
-                secretKeyRef: 
-                  name: kong-admin-auth-apikey
-                  key: value
-            - name: KONG_ADMIN_AUTH_APIKEY_HEADER
-              valueFrom:
-                secretKeyRef: 
-                  name: kong-admin-auth-apikey
-                  key: header
-            {{- end }}
+            - name: KONG_LOGS_HTTP_SERVER_PATH
+              value: "{{ .Values.kong.logs.http.path }}"
+            - name: KONG_LOGS_HTTP_SERVER_PORT
+              value: "{{ .Values.kong.logs.http.port }}"
           {{- with .Values.env }}
           {{- range $key, $value := . }}
           {{- if and (not (eq (toString $value) ""))
@@ -92,19 +85,36 @@ spec:
                                                      "KONG_PROXY_PORTS_HTTP" 
                                                      "KONG_PROXY_PORTS_HTTPS" 
                                                      "KONG_SPEC_LOCALPATH"
-                                                     "KONG_SPEC_URLPATHS")))
+                                                     "KONG_SPEC_URLPATHS"
+                                                     "KONG_LOGS_HTTP_SERVER_PATH"
+                                                     "KONG_LOGS_HTTP_SERVER_PORT"
+                                                     "STATUS_PORT")))
            }}
             - name: {{ $key }}
               value: {{ $value | quote }} 
           {{- end }}
           {{- end }}
           {{- end }}
+            - name: CENTRAL_AUTH_PRIVATEKEY
+              value: "/keys/private_key.pem"
+            - name: CENTRAL_AUTH_PUBLICKEY
+              value: "/keys/public_key.pem"
+            - name: STATUS_PORT
+              value: "{{ .Values.statusPort }}"
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
           volumeMounts:
+            - name: beat-storage
+              mountPath: /data
             - name: "kong-agent-keys"
               mountPath: "/keys"
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
       volumes:
+      - name: beat-storage
+        emptyDir: {}
       - name: kong-agent-keys
         secret:
           secretName: {{ .Values.secrets.keys }}

helm/kong-agents/values.yaml

@@ -41,6 +41,10 @@ kong:
     filter:
     urlPaths: []
     localPath:
+  logs:
+    http:
+      path: /requestlogs
+      port: 9000
 
 # Add any environment variable overrides here
 env: {}

pkg/kong/provisioning.go

@@ -360,15 +360,15 @@ func getSpecificPlugin(plugins []*klib.Plugin, serviceID, routeID, consumerID, p
 			continue
 		}
 
-		if consumerID == "" || plugin.Consumer == nil || (plugin.Consumer != nil && *plugin.Consumer.ID == consumerID) {
+		if (consumerID == "" && plugin.Consumer == nil) || (consumerID != "" && plugin.Consumer != nil && *plugin.Consumer.ID == consumerID) {
 			consumerMatch = true
 		}
 
-		if routeID == "" || plugin.Route == nil || (plugin.Route != nil && *plugin.Route.ID == routeID) {
+		if (routeID == "" && plugin.Route == nil) || (routeID != "" && plugin.Route != nil && *plugin.Route.ID == routeID) {
 			routeMatch = true
 		}
 
-		if serviceID == "" || plugin.Service == nil || (plugin.Service != nil && *plugin.Service.ID == serviceID) {
+		if (serviceID == "" && plugin.Service == nil) || (serviceID != "" && plugin.Service != nil && *plugin.Service.ID == serviceID) {
 			serviceMatch = true
 		}
 

pkg/kong/provisioning_test.go

@@ -584,6 +584,9 @@ func TestAddQuota(t *testing.T) {
 									ID: klib.String("routeID"),
 								},
 								Enabled: klib.Bool(true),
+								Consumer: &klib.Consumer{
+									ID: klib.String("consumerID"),
+								},
 							},
 						},
 						"next": "null",
@@ -606,6 +609,9 @@ func TestAddQuota(t *testing.T) {
 								Route: &klib.Route{
 									ID: klib.String("routeID"),
 								},
+								Consumer: &klib.Consumer{
+									ID: klib.String("consumerID"),
+								},
 								Enabled: klib.Bool(false),
 							},
 						},

pkg/traceability/beater/beater.go

@@ -5,11 +5,15 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"os"
+	"sync"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/google/uuid"
 
+	"github.com/Axway/agent-sdk/pkg/agent"
+	"github.com/Axway/agent-sdk/pkg/transaction/metric"
 	agentErrors "github.com/Axway/agent-sdk/pkg/util/errors"
 	hc "github.com/Axway/agent-sdk/pkg/util/healthcheck"
 	"github.com/Axway/agent-sdk/pkg/util/log"
@@ -19,15 +23,19 @@ import (
 )
 
 type httpLogBeater struct {
-	client beat.Client
-	logger log.FieldLogger
-	server http.Server
+	client       beat.Client
+	logger       log.FieldLogger
+	server       http.Server
+	processing   sync.WaitGroup
+	shutdownDone sync.WaitGroup
 }
 
 // New creates an instance of kong_traceability_agent.
 func New(*beat.Beat, *common.Config) (beat.Beater, error) {
 	b := &httpLogBeater{
-		logger: log.NewFieldLogger().WithComponent("httpLogBeater").WithPackage("beater"),
+		logger:       log.NewFieldLogger().WithComponent("httpLogBeater").WithPackage("beater"),
+		processing:   sync.WaitGroup{},
+		shutdownDone: sync.WaitGroup{},
 	}
 
 	// Validate that all necessary services are up and running. If not, return error
@@ -48,18 +56,23 @@ func (b *httpLogBeater) Run(beater *beat.Beat) error {
 	if err != nil {
 		return err
 	}
+	agent.RegisterShutdownHandler(b.shutdownHandler)
 
 	mux := http.NewServeMux()
-	mux.HandleFunc(config.GetAgentConfig().HttpLogPluginConfig.Path, b.HandleHello)
+	mux.HandleFunc(config.GetAgentConfig().KongGatewayCfg.Logs.HTTP.Path, b.HandleEvent)
 
 	// other handlers can be assigned to separate paths
-	b.server = http.Server{Handler: mux, Addr: fmt.Sprintf(":%d", config.GetAgentConfig().HttpLogPluginConfig.Port)}
-	b.logger.Fatal(b.server.ListenAndServe())
+	b.server = http.Server{Handler: mux, Addr: fmt.Sprintf(":%d", config.GetAgentConfig().KongGatewayCfg.Logs.HTTP.Port)}
+	b.server.ListenAndServe()
+
+	// wait for the shutdown process to finish prior to exit
+	b.shutdownDone.Add(1)
+	b.shutdownDone.Wait()
 
 	return nil
 }
 
-func (b *httpLogBeater) HandleHello(w http.ResponseWriter, r *http.Request) {
+func (b *httpLogBeater) HandleEvent(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		b.logger.Trace("received a non post request")
 		w.WriteHeader(http.StatusMethodNotAllowed)
@@ -74,19 +87,37 @@ func (b *httpLogBeater) HandleHello(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	b.processing.Add(1)
 	go func(data []byte) {
+		defer b.processing.Done()
 		ctx := context.WithValue(context.Background(), processor.CtxTransactionID, uuid.NewString())
 
 		eventProcessor, err := processor.NewEventsHandler(ctx, data)
 		if err == nil {
-			eventsToPublish, err := eventProcessor.Handle()
-			if err == nil {
-				b.client.PublishAll(eventsToPublish)
-			}
+			eventsToPublish := eventProcessor.Handle()
+			b.client.PublishAll(eventsToPublish)
 		}
 	}(logData)
 }
 
+func (b *httpLogBeater) shutdownHandler() {
+	b.logger.Info("waiting for current processing to finish")
+	defer b.shutdownDone.Done()
+
+	// wait for all processing to finish
+	b.processing.Wait()
+
+	// publish the metrics and usage
+	b.logger.Info("publishing cached metrics and usage")
+	metric.GetMetricCollector().ShutdownPublish()
+
+	// remove the agent resource in k8s clusters
+	pod_name := os.Getenv("POD_NAME")
+	if pod_name != "" {
+		agent.GetCentralClient().DeleteResourceInstance(agent.GetAgentResource())
+	}
+}
+
 // Stop stops kong_traceability_agent.
 func (b *httpLogBeater) Stop() {
 	b.server.Shutdown(context.Background())

pkg/traceability/cmd/cmd.go

@@ -50,8 +50,8 @@ func initConfig(centralConfig corecfg.CentralConfig) (interface{}, error) {
 	rootProps := TraceCmd.GetProperties()
 
 	agentConfig := &config.AgentConfig{
-		CentralCfg:          centralConfig,
-		HttpLogPluginConfig: config.ParseProperties(rootProps),
+		CentralCfg:     centralConfig,
+		KongGatewayCfg: config.ParseProperties(rootProps),
 	}
 
 	config.SetAgentConfig(agentConfig)