Add Kubernetes Infra transform rule

netobserv · Dec 12, 2023 · 0d23193 · 0d23193
1 parent 3d2898b
commit 0d23193
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 0 deletions.
diff --git a/docs/api.md b/docs/api.md
@@ -178,6 +178,7 @@ Following is the supported API format for network transformations:
                      add_location: add output location fields from input
                      add_service: add output network service field from input port and parameters protocol field
                      add_kubernetes: add output kubernetes fields from input
+                     add_kubernetes_infra: add output kubernetes isInfra field from input
                      reinterpret_direction: reinterpret flow direction at the node level (instead of net interface), to ease the deduplication process
                      add_ip_category: categorize IPs based on known subnets configuration
                  parameters: parameters specific to type

diff --git a/pkg/api/api.go b/pkg/api/api.go
@@ -45,6 +45,7 @@ const (
 	AddLocationRuleType          = "add_location"
 	AddServiceRuleType           = "add_service"
 	AddKubernetesRuleType        = "add_kubernetes"
+	AddKubernetesInfraRuleType   = "add_kubernetes_infra"
 	ReinterpretDirectionRuleType = "reinterpret_direction"
 	PromFilterExact              = "exact"
 	PromFilterPresence           = "presence"

diff --git a/pkg/api/transform_network.go b/pkg/api/transform_network.go
@@ -43,6 +43,7 @@ const (
 	OpAddLocation          = "add_location"
 	OpAddService           = "add_service"
 	OpAddKubernetes        = "add_kubernetes"
+	OpAddKubernetesInfra   = "add_kubernetes_infra"
 	OpReinterpretDirection = "reinterpret_direction"
 	OpAddIPCategory        = "add_ip_category"
 )
@@ -52,6 +53,7 @@ type TransformNetworkOperationEnum struct {
 	AddLocation          string `yaml:"add_location" json:"add_location" doc:"add output location fields from input"`
 	AddService           string `yaml:"add_service" json:"add_service" doc:"add output network service field from input port and parameters protocol field"`
 	AddKubernetes        string `yaml:"add_kubernetes" json:"add_kubernetes" doc:"add output kubernetes fields from input"`
+	AddKubernetesInfra   string `yaml:"add_kubernetes_infra" json:"add_kubernetes_infra" doc:"add output kubernetes isInfra field from input"`
 	ReinterpretDirection string `yaml:"reinterpret_direction" json:"reinterpret_direction" doc:"reinterpret flow direction at the node level (instead of net interface), to ease the deduplication process"`
 	AddIPCategory        string `yaml:"add_ip_category" json:"add_ip_category" doc:"categorize IPs based on known subnets configuration"`
 }

diff --git a/pkg/pipeline/transform/transform_network.go b/pkg/pipeline/transform/transform_network.go
@@ -123,6 +123,24 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 					outputEntry[rule.Output+"_HostName"] = kubeInfo.HostName
 				}
 			}
+		case api.OpAddKubernetesInfra:
+			srcKubeInfo, err := kubernetes.Data.GetInfo(fmt.Sprintf("%s", outputEntry["SrcAddr"]))
+			if err != nil {
+				logrus.WithError(err).Tracef("can't find kubernetes info for IP %v", outputEntry["SrcAddr"])
+				continue
+			}
+			dstKubeInfo, err := kubernetes.Data.GetInfo(fmt.Sprintf("%s", outputEntry["DstAddr"]))
+			if err != nil {
+				logrus.WithError(err).Tracef("can't find kubernetes info for IP %v", outputEntry["DstAddr"])
+				continue
+			}
+
+			if objectIsApp(*srcKubeInfo, rule.Parameters) || objectIsApp(*dstKubeInfo, rule.Parameters) {
+				outputEntry["K8S_FlowType"] = "app"
+			} else {
+				outputEntry["K8S_FlowType"] = "infra"
+			}
+
 		case api.OpReinterpretDirection:
 			reinterpretDirection(outputEntry, &n.DirectionInfo)
 		case api.OpAddIPCategory:
@@ -143,6 +161,28 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 	return outputEntry, true
 }
 
+const openshiftNamespacePrefix = "openshift-"
+const openshiftPrefixLen = len(openshiftNamespacePrefix)
+
+func objectIsApp(obj kubernetes.Info, additionalInfraPrefix string) bool {
+	nsLen := len(obj.Namespace)
+	additionalPrefixLen := len(additionalInfraPrefix)
+	if nsLen == 0 {
+		return false
+	}
+	if nsLen >= openshiftPrefixLen && obj.Namespace[:openshiftPrefixLen] == openshiftNamespacePrefix {
+		return false
+	}
+	if nsLen >= additionalPrefixLen && obj.Namespace[:additionalPrefixLen] == additionalInfraPrefix {
+		return false
+	}
+	//Special case with openshift and kubernetes service in default namespace
+	if obj.Namespace == "default" && (obj.Name == "kubernetes" || obj.Name == "openshift") {
+		return false
+	}
+	return true
+}
+
 func (n *Network) categorizeIP(ip net.IP) string {
 	if ip != nil {
 		for _, subnetCat := range n.categories {
@@ -172,6 +212,8 @@ func NewTransformNetwork(params config.StageParam) (Transformer, error) {
 			needToInitLocationDB = true
 		case api.OpAddKubernetes:
 			needToInitKubeData = true
+		case api.OpAddKubernetesInfra:
+			needToInitKubeData = true
 		case api.OpAddService:
 			needToInitNetworkServices = true
 		case api.OpReinterpretDirection: