update sentryflow

Signed-off-by: Jaehyun Nam <[email protected]>

sentryflow/Dockerfile

@@ -21,6 +21,8 @@ RUN make build
 WORKDIR /app
 COPY /sentryflow .
 
+RUN go install github.com/golang/protobuf/protoc-gen-go@latest
+RUN go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
 RUN go mod tidy
 RUN export CGO_ENABLED=1; export CC=gcc;
 RUN go build -o sentryflow

sentryflow/Makefile

@@ -1,30 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
-IMAGE_NAME = 5gsec/sentryflow
+PROG_NAME = sentryflow
+IMAGE_NAME = 5gsec/$(PROG_NAME)
 TAG = v0.1
 
-
-.PHONY: build
-build:
-	go mod tidy
-	go build -o sentryflow
-
-.PHONY: image
-image:
-	docker build -t $(IMAGE_NAME):$(TAG) -f ./Dockerfile ../
-
-.PHONY: clean-build
-clean-build:
-	rm -f sentryflow
-
-.PHONY: clean-image
-clean-image:
-	docker rmi $(IMAGE_NAME):$(TAG)
-
-.PHONY: run
-run:
-	docker run -it --rm $(IMAGE_NAME):$(TAG)
-
 .PHONY: gofmt
 gofmt:
 	cd $(CURDIR); gofmt -w -s -d $(shell find . -type f -name '*.go' -print)
@@ -58,3 +37,24 @@ ifeq (, $(shell which gosec))
 	}
 endif
 	cd $(CURDIR); gosec -exclude=G402 ./...
+
+.PHONY: build
+build:
+	go mod tidy
+	go build -o $(PROG_NAME)
+
+.PHONY: clean
+clean:
+	rm -f $(PROG_NAME)
+
+.PHONY: build-image
+build-image:
+	docker build -t $(IMAGE_NAME):$(TAG) -f ./Dockerfile ../
+
+.PHONY: clean-image
+clean-image:
+	docker rmi $(IMAGE_NAME):$(TAG)
+
+.PHONY: run-image
+run-image:
+	docker run -it --rm $(IMAGE_NAME):$(TAG)

sentryflow/collector/collectorHandler.go

@@ -5,7 +5,7 @@ package collector
 import (
 	"errors"
 	"fmt"
-	cfg "github.com/5GSEC/sentryflow/config"
+	cfg "github.com/5GSEC/SentryFlow/config"
 	"google.golang.org/grpc"
 	"log"
 	"net"

sentryflow/collector/envoy.go

@@ -3,14 +3,13 @@
 package collector
 
 import (
-	"fmt"
-	"github.com/5GSEC/sentryflow/core"
-	"github.com/5GSEC/sentryflow/protobuf"
+	"io"
+	"log"
+
+	"github.com/5GSEC/SentryFlow/core"
 	envoyAls "github.com/envoyproxy/go-control-plane/envoy/service/accesslog/v3"
 	envoyMetrics "github.com/envoyproxy/go-control-plane/envoy/service/metrics/v3"
 	"google.golang.org/grpc"
-	"io"
-	"log"
 )
 
 // EnvoyMetricsServer Structure
@@ -52,30 +51,9 @@ func (ems *EnvoyMetricsServer) StreamMetrics(stream envoyMetrics.MetricsService_
 
 	if identifier != nil {
 		log.Printf("[Envoy] Received EnvoyMetric - ID: %s, %s", identifier.GetNode().GetId(), identifier.GetNode().GetCluster())
+		metaData := identifier.GetNode().GetMetadata().AsMap()
 
-		nodeID := identifier.GetNode().GetId()
-		cluster := identifier.GetNode().GetCluster()
-
-		curIdentifier := fmt.Sprintf("%s, %s", nodeID, cluster)
-		envoyMetric := &protobuf.EnvoyMetric{
-			Identifier: curIdentifier,
-			Metric:     []*protobuf.Metric{},
-		}
-
-		for _, metric := range event.GetEnvoyMetrics() {
-			metricType := metric.GetType().String()
-			metricName := metric.GetName()
-			tempMetrics := metric.GetMetric()
-			metrics := fmt.Sprintf("%s", tempMetrics)
-
-			curMetric := &protobuf.Metric{
-				Type:  metricType,
-				Key:   metricName,
-				Value: metrics,
-			}
-
-			envoyMetric.Metric = append(envoyMetric.Metric, curMetric)
-		}
+		envoyMetric := core.GenerateMetricFromEnvoy(event, metaData)
 
 		core.Lh.InsertLog(envoyMetric)
 	}

sentryflow/collector/opentelemetry.go

@@ -4,7 +4,7 @@ package collector
 
 import (
 	"context"
-	"github.com/5GSEC/sentryflow/core"
+	"github.com/5GSEC/SentryFlow/core"
 	otelLogs "go.opentelemetry.io/proto/otlp/collector/logs/v1"
 	"google.golang.org/grpc"
 )

sentryflow/config/config.go

@@ -23,10 +23,14 @@ type SentryFlowConfig struct {
 	PatchNamespace          bool // Enable/Disable patching namespace for Istio injection
 	PatchRestartDeployments bool // Enable/Disable restarting deployments after patching
 
-	AIEngineService   string
-	AIEngineBatchSize int
+	AIEngineService     string
+	AIEngineServicePort string
+	AIEngineBatchSize   int
 
-	MetricsDBFileName string // String value of MetricsDB file (sqlite3 db file)
+	MetricsDBFileName        string // String value of MetricsDB file (sqlite3 db file)
+	MetricsDBAggregationTime int    // Value of APILog Aggregation Time
+	MetricsDBClearTime       int    // Value of APIMetric DB Clear time
+	APIMetricsSendTime       int    // Value of APIMetric send time
 
 	CollectorEnableOpenTelemetry bool // Enable/Disable OpenTelemetry Collector
 	Debug                        bool // Enable/Disable SentryFlow debug mode
@@ -49,8 +53,12 @@ const (
 	PatchNamespace               string = "patchNamespace"
 	PatchRestartDeployments      string = "patchRestartDeployments"
 	AIEngineService              string = "aiEngineService"
+	AIEngineServicePort          string = "aiEngineServicePort"
 	AIEngineBatchSize            string = "aiEngineBatchSize"
 	MetricsDBFileName            string = "metricsDBFileName"
+	MetricsDBAggregationTime     string = "metricsDBAggregationTime"
+	MetricsDBClearTime           string = "metricsDBClearTime"
+	APIMetricsSendTime           string = "apiMetricsSendTime"
 	CollectorEnableOpenTelemetry string = "collectorEnableOpenTelemetry"
 	Debug                        string = "debug"
 )
@@ -63,8 +71,12 @@ func readCmdLineParams() {
 	patchNamespaceB := flag.Bool(PatchNamespace, false, "Enable/Disable patching Istio injection to all namespaces")
 	patchRestartDeploymentsB := flag.Bool(PatchRestartDeployments, false, "Enable/Disable restarting deployments in all namespaces")
 	aiEngineServiceStr := flag.String(AIEngineService, "ai-engine.sentryflow.svc.cluster.local", "Service address for SentryFlow AI Engine")
-	aiEngineBatchSizeInt := flag.Int(AIEngineBatchSize, 5, "Batch size fo SentryFlow AI Engine")
+	aiEngineServicePortStr := flag.String(AIEngineServicePort, "5000", "Service Port for SentryFlow AI Engine")
+	aiEngineBatchSizeInt := flag.Int(AIEngineBatchSize, 5, "Batch size for SentryFlow AI Engine")
 	metricsDBFileNameStr := flag.String(MetricsDBFileName, "/etc/sentryflow/metrics.db", "File name for local metrics DB")
+	metricsDBAggregationTimeInt := flag.Int(MetricsDBAggregationTime, 10, "Term time between aggregations")
+	metricsDBClearTimeInt := flag.Int(MetricsDBClearTime, 600, "Metrics DB Clear Time")
+	APIMetricsSendTimeInt := flag.Int(APIMetricsSendTime, 10, "APIMetric send term")
 	collectorEnableOpenTelemetryB := flag.Bool(CollectorEnableOpenTelemetry, true, "Enable/Disable OpenTelemetry Collector")
 	configDebugB := flag.Bool(Debug, false, "Enable/Disable debugging mode using logs")
 
@@ -84,8 +96,12 @@ func readCmdLineParams() {
 	viper.SetDefault(PatchNamespace, *patchNamespaceB)
 	viper.SetDefault(PatchRestartDeployments, *patchRestartDeploymentsB)
 	viper.SetDefault(AIEngineService, *aiEngineServiceStr)
+	viper.SetDefault(AIEngineServicePort, *aiEngineServicePortStr)
 	viper.SetDefault(AIEngineBatchSize, *aiEngineBatchSizeInt)
 	viper.SetDefault(MetricsDBFileName, *metricsDBFileNameStr)
+	viper.SetDefault(MetricsDBAggregationTime, *metricsDBAggregationTimeInt)
+	viper.SetDefault(MetricsDBClearTime, *metricsDBClearTimeInt)
+	viper.SetDefault(APIMetricsSendTime, *APIMetricsSendTimeInt)
 	viper.SetDefault(CollectorEnableOpenTelemetry, *collectorEnableOpenTelemetryB)
 	viper.SetDefault(Debug, *configDebugB)
 }
@@ -108,8 +124,12 @@ func LoadConfig() error {
 	GlobalCfg.PatchNamespace = viper.GetBool(PatchNamespace)
 	GlobalCfg.PatchRestartDeployments = viper.GetBool(PatchRestartDeployments)
 	GlobalCfg.AIEngineService = viper.GetString(AIEngineService)
+	GlobalCfg.AIEngineServicePort = viper.GetString(AIEngineServicePort)
 	GlobalCfg.AIEngineBatchSize = viper.GetInt(AIEngineBatchSize)
 	GlobalCfg.MetricsDBFileName = viper.GetString(MetricsDBFileName)
+	GlobalCfg.MetricsDBAggregationTime = viper.GetInt(MetricsDBAggregationTime)
+	GlobalCfg.MetricsDBClearTime = viper.GetInt(MetricsDBClearTime)
+	GlobalCfg.APIMetricsSendTime = viper.GetInt(APIMetricsSendTime)
 	GlobalCfg.CollectorEnableOpenTelemetry = viper.GetBool(CollectorEnableOpenTelemetry)
 	GlobalCfg.Debug = viper.GetBool(Debug)
 

sentryflow/core/k8sHandler.go

@@ -4,18 +4,19 @@ package core
 
 import (
 	"context"
-	"github.com/5GSEC/sentryflow/config"
-	"github.com/5GSEC/sentryflow/types"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/5GSEC/SentryFlow/config"
+	"github.com/5GSEC/SentryFlow/types"
 	"gopkg.in/yaml.v2"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
-	"log"
-	"sync"
-	"time"
 )
 
 // K8s global reference for Kubernetes Handler
@@ -329,7 +330,6 @@ func (kh *K8sHandler) PatchIstioConfigMap() error {
 	}
 
 	_, eeaExist := meshConfig["enableEnvoyAccessLogService"]
-
 	if eeaExist {
 		log.Printf("Overwrite the contents of \"enableEnvoyAccessLogService\"")
 	}

sentryflow/core/logHandler.go

@@ -3,15 +3,17 @@
 package core
 
 import (
-	"github.com/5GSEC/sentryflow/exporter"
-	"github.com/5GSEC/sentryflow/metrics"
-	"github.com/5GSEC/sentryflow/protobuf"
-	"github.com/5GSEC/sentryflow/types"
-	accesslogv3 "github.com/envoyproxy/go-control-plane/envoy/data/accesslog/v3"
 	"log"
 	"strconv"
 	"strings"
 	"sync"
+
+	"github.com/5GSEC/SentryFlow/exporter"
+	"github.com/5GSEC/SentryFlow/metrics"
+	"github.com/5GSEC/SentryFlow/protobuf"
+	"github.com/5GSEC/SentryFlow/types"
+	accesslogv3 "github.com/envoyproxy/go-control-plane/envoy/data/accesslog/v3"
+	metricv3 "github.com/envoyproxy/go-control-plane/envoy/service/metrics/v3"
 )
 
 // Lh global reference for LogHandler
@@ -28,6 +30,13 @@ type LogHandler struct {
 	logChan  chan interface{}
 }
 
+// aggregationLog Structure
+type aggregationLog struct {
+	Logs        []*protobuf.APILog
+	Labels      map[string]string
+	Annotations map[string]string
+}
+
 // NewLogHandler Structure
 func NewLogHandler() *LogHandler {
 	lh := &LogHandler{
@@ -87,6 +96,7 @@ func processAccessLog(al *protobuf.APILog) {
 	metrics.InsertAccessLog(al)
 }
 
+// processEnvoyMetric Function
 func processEnvoyMetric(em *protobuf.EnvoyMetric) {
 	exporter.InsertEnvoyMetric(em)
 }
@@ -228,3 +238,62 @@ func GenerateAccessLogsFromEnvoy(entry *accesslogv3.HTTPAccessLogEntry) *protobu
 
 	return envoyAccessLog
 }
+
+// GenerateMetricFromEnvoy Function
+func GenerateMetricFromEnvoy(event *metricv3.StreamMetricsMessage, metaData map[string]interface{}) *protobuf.EnvoyMetric {
+	pod := LookupNetworkedResource(metaData["INSTANCE_IPS"].(string))
+	envoyMetric := &protobuf.EnvoyMetric{
+		PodIP:     metaData["INSTANCE_IPS"].(string),
+		Name:      metaData["NAME"].(string),
+		Namespace: metaData["NAMESPACE"].(string),
+		Labels:    pod.Labels,
+		TimeStamp: "",
+		Metric:    make(map[string]*protobuf.MetricValue),
+	}
+
+	envoyMetric.Metric["GAUGE"] = &protobuf.MetricValue{
+		Value: make(map[string]string),
+	}
+	envoyMetric.Metric["COUNTER"] = &protobuf.MetricValue{
+		Value: make(map[string]string),
+	}
+	envoyMetric.Metric["HISTOGRAM"] = &protobuf.MetricValue{
+		Value: make(map[string]string),
+	}
+	envoyMetric.Metric["SUMMARY"] = &protobuf.MetricValue{
+		Value: make(map[string]string),
+	}
+
+	for _, metric := range event.GetEnvoyMetrics() {
+		metricType := metric.GetType().String()
+		metricName := metric.GetName()
+
+		if envoyMetric.Metric[metricType].Value == nil {
+			continue
+		}
+
+		var metricValue string
+
+		for _, metricDetail := range metric.GetMetric() {
+			if envoyMetric.TimeStamp == "" {
+				envoyMetric.TimeStamp = strconv.FormatInt(metricDetail.GetTimestampMs(), 10)
+			}
+			if metricType == "GAUGE" {
+				metricValue = strconv.FormatFloat(metricDetail.GetGauge().GetValue(), 'f', -1, 64)
+			}
+			if metricType == "COUNTER" {
+				metricValue = strconv.FormatFloat(metricDetail.GetCounter().GetValue(), 'f', -1, 64)
+			}
+			if metricType == "HISTOGRAM" {
+				metricValue = strconv.FormatUint(metricDetail.GetHistogram().GetSampleCount(), 10)
+			}
+			if metricType == "SUMMARY" {
+				metricValue = strconv.FormatUint(metricDetail.GetHistogram().GetSampleCount(), 10)
+			}
+
+			envoyMetric.Metric[metricType].Value[metricName] = metricValue
+		}
+	}
+
+	return envoyMetric
+}