Repository: skupperproject/skupper
Pull request #1243 (2023-09-29 16:59:53)

add envars to emit complete flow records
To reconcile proxies properly in the collector. Also check whether host is ip address or hostname in order to correlate to connectors.


Patch:
diff --git a/pkg/flow/flow_mem_driver.go b/pkg/flow/flow_mem_driver.go
index 9f8206ac..e67eb1c8 100644
--- a/pkg/flow/flow_mem_driver.go
+++ b/pkg/flow/flow_mem_driver.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"net"
 	"strings"
 	"time"
 
@@ -755,8 +756,8 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 			} else {
 				if connector.EndTime > 0 {
 					current.EndTime = connector.EndTime
-					if current.process != nil {
-						if process, ok := fc.Processes[*current.process]; ok {
+					if current.ProcessId != nil {
+						if process, ok := fc.Processes[*current.ProcessId]; ok {
 							process.connector = nil
 							process.ProcessBinding = &Unbound
 						}
@@ -1337,8 +1338,8 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "process":
 			if id, ok := vars["id"]; ok {
 				if connector, ok := fc.Connectors[id]; ok {
-					if connector.process != nil {
-						if process, ok := fc.Processes[*connector.process]; ok {
+					if connector.ProcessId != nil {
+						if process, ok := fc.Processes[*connector.ProcessId]; ok {
 							p.Count = 1
 							p.Results = *process
 						}
@@ -1448,8 +1449,8 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			if id, ok := vars["id"]; ok {
 				if vanaddr, ok := fc.VanAddresses[id]; ok {
 					for _, connector := range fc.Connectors {
-						if *connector.Address == vanaddr.Name && connector.process != nil {
-							if process, ok := fc.Processes[*connector.process]; ok {
+						if *connector.Address == vanaddr.Name && connector.ProcessId != nil {
+							if process, ok := fc.Processes[*connector.ProcessId]; ok {
 								if filterRecord(*process, queryParams) && process.Base.TimeRangeValid(queryParams) {
 									unique[process.Identity] = process
 								}
@@ -1468,10 +1469,10 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			if id, ok := vars["id"]; ok {
 				if vanaddr, ok := fc.VanAddresses[id]; ok {
 					for _, connector := range fc.Connectors {
-						if *connector.Address == vanaddr.Name && connector.process != nil {
+						if *connector.Address == vanaddr.Name && connector.ProcessId != nil {
 							for _, aggregate := range fc.FlowAggregates {
 								if aggregate.PairType == recordNames[Process] {
-									if *connector.process == *aggregate.DestinationId {
+									if *connector.ProcessId == *aggregate.DestinationId {
 										p.TotalCount++
 										if filterRecord(*aggregate, queryParams) {
 											processPairs = append(processPairs, *aggregate)
@@ -1551,7 +1552,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			if id, ok := vars["id"]; ok {
 				if _, ok := fc.Processes[id]; ok {
 					for _, connector := range fc.Connectors {
-						if connector.process != nil && *connector.process == id {
+						if connector.ProcessId != nil && *connector.ProcessId == id {
 							for _, address := range fc.VanAddresses {
 								if *connector.Address == address.Name {
 									if filterRecord(*address, queryParams) {
@@ -2070,8 +2071,8 @@ func (fc *FlowCollector) reconcileRecords() error {
 			if flow.SourceHost != nil {
 				if connector, ok := fc.Connectors[flow.Parent]; ok {
 					flow.Place = serverSide
-					if connector.process != nil && connector.AddressId != nil {
-						flow.Process = connector.process
+					if connector.ProcessId != nil && connector.AddressId != nil {
+						flow.Process = connector.ProcessId
 						if process, ok := fc.Processes[*flow.Process]; ok {
 							flow.ProcessName = process.Name
 						}
@@ -2144,12 +2145,21 @@ func (fc *FlowCollector) reconcileRecords() error {
 				delete(fc.connectorsToReconcile, connId)
 			} else if connector.DestHost != nil {
 				siteId := fc.getRecordSiteId(*connector)
+				var matchHost *string
 				found := false
+				if net.ParseIP(*connector.DestHost) == nil {
+					addrs, err := net.LookupHost(*connector.DestHost)
+					if err == nil && len(addrs) > 0 {
+						matchHost = &addrs[0]
+					}
+				} else {
+					matchHost = connector.DestHost
+				}
 				for _, process := range fc.Processes {
 					if siteId == process.Parent {
 						if process.SourceHost != nil {
-							if *connector.DestHost == *process.SourceHost {
-								connector.process = &process.Identity
+							if *matchHost == *process.SourceHost {
+								connector.ProcessId = &process.Identity
 								process.connector = &connector.Identity
 								process.ProcessBinding = &Bound
 								log.Printf("COLLECTOR: Connector %s/%s associated to process %s\n", connector.Identity, *connector.Address, *process.Name)
@@ -2174,7 +2184,7 @@ func (fc *FlowCollector) reconcileRecords() error {
 						for _, process := range fc.Processes {
 							if process.Name != nil && *process.Name == processName {
 								log.Printf("COLLECTOR: Associating connector %s to external process %s\n", connector.Identity, processName)
-								connector.process = &process.Identity
+								connector.ProcessId = &process.Identity
 								delete(fc.connectorsToReconcile, connId)
 								break
 							}
diff --git a/pkg/flow/record.go b/pkg/flow/record.go
index a1b26dbd..6ac887b2 100644
--- a/pkg/flow/record.go
+++ b/pkg/flow/record.go
@@ -283,7 +283,7 @@ type ConnectorRecord struct {
 	FlowRateL4  *uint64 `json:"flowRateL4,omitempty"`
 	FlowCountL7 *uint64 `json:"flowCountL7,omitempty"`
 	FlowRateL7  *uint64 `json:"flowRateL7,omitempty"`
-	process     *string
+	ProcessId   *string `json:"processId,omitempty"`
 	AddressId   *string `json:"addressId,omitempty"`
 }
 
diff --git a/pkg/kube/deployments.go b/pkg/kube/deployments.go
index 360100da..f2a3b4b3 100644
--- a/pkg/kube/deployments.go
+++ b/pkg/kube/deployments.go
@@ -3,6 +3,7 @@ package kube
 import (
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	"github.com/skupperproject/skupper/pkg/qdr"
@@ -129,7 +130,10 @@ func NewProxyStatefulSet(image types.ImageDetails, serviceInterface types.Servic
 
 	replicas := int32(serviceInterface.Headless.Size)
 	labels := map[string]string{
-		"internal.skupper.io/type": "proxy",
+		"internal.skupper.io/type":  "proxy",
+		"app.kubernetes.io/name":    "skupper-router",
+		"app.kubernetes.io/part-of": "skupper",
+		"skupper.io/component":      "router",
 	}
 	if len(transportDep.ObjectMeta.Labels) > 0 {
 		for k, v := range transportDep.ObjectMeta.Labels {
@@ -168,6 +172,9 @@ func NewProxyStatefulSet(image types.ImageDetails, serviceInterface types.Servic
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
 						"internal.skupper.io/service": serviceInterface.Address,
+						"app.kubernetes.io/name":      "skupper-router",
+						"app.kubernetes.io/part-of":   "skupper",
+						"skupper.io/component":        "router",
 					},
 				},
 				Spec: corev1.PodSpec{
@@ -187,7 +194,11 @@ func NewProxyStatefulSet(image types.ImageDetails, serviceInterface types.Servic
 									Value: "json",
 								},
 								{
-									Name: "NAMESPACE",
+									Name:  "SKUPPER_SITE_ID",
+									Value: os.Getenv("SKUPPER_SITE_ID"),
+								},
+								{
+									Name: "POD_NAMESPACE",
 									ValueFrom: &corev1.EnvVarSource{
 										FieldRef: &corev1.ObjectFieldSelector{
 											FieldPath: "metadata.namespace",


Test patch:
diff --git a/pkg/flow/flow_mem_driver_test.go b/pkg/flow/flow_mem_driver_test.go
index 22d6e6b3..1a1381e1 100644
--- a/pkg/flow/flow_mem_driver_test.go
+++ b/pkg/flow/flow_mem_driver_test.go
@@ -180,9 +180,9 @@ func TestRecordGraph(t *testing.T) {
 				Parent:    "router:0",
 				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
 			},
-			Address:  &address3,
-			Protocol: &protocol,
-			process:  &connectorProcess1,
+			Address:   &address3,
+			Protocol:  &protocol,
+			ProcessId: &connectorProcess1,
 		},
 		{
 			Base: Base{
@@ -200,9 +200,9 @@ func TestRecordGraph(t *testing.T) {
 				Parent:    "router:1",
 				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
 			},
-			Protocol: &protocol,
-			Address:  &address4,
-			process:  &connectorProcess2,
+			Protocol:  &protocol,
+			Address:   &address4,
+			ProcessId: &connectorProcess2,
 		},
 	}
 	flows := []FlowRecord{