Merge branch 'main' of github.com:kurtosis-tech/kardinal-kontrol

kontrol-service/api/analytics.go

@@ -19,16 +19,13 @@ const (
 )
 
 // NewAnalyticsWrapper creates a new AnalyticsWrapper
-func NewAnalyticsWrapper(isDevMode bool) *AnalyticsWrapper {
-	if !isDevMode {
-		// This is the Segment write key for the "kontrol-service" project. It is not
-		// a sensitive value, but it could be extracted to an env var in the future
-		// to separate dev and prod traffic if desired.
-		client := analytics.New("1TeZVRY3ta9VYaNknKTCCKBZtcBllE6U")
+func NewAnalyticsWrapper(isDevMode bool, writeKey string) *AnalyticsWrapper {
+	if !isDevMode && writeKey != "" {
+		client := analytics.New(writeKey)
 		logrus.Info("Segment analytics client initialized")
 		return &AnalyticsWrapper{client: &client}
 	}
-	logrus.Info("Dev mode: Segment analytics client not initialized")
+	logrus.Info("Dev mode or write key not set: Segment analytics client not initialized")
 	return &AnalyticsWrapper{client: nil}
 }
 

kontrol-service/engine/flow/merger.go

@@ -1,13 +1,14 @@
 package flow
 
 import (
+	"encoding/json"
 	"github.com/samber/lo"
 	v1 "k8s.io/api/networking/v1"
 	"kardinal.kontrol-service/types/cluster_topology/resolved"
 )
 
-func MergeClusterTopologies(baseTopology resolved.ClusterTopology, clusterTopologies []resolved.ClusterTopology) *resolved.ClusterTopology {
-	mergedTopology := &resolved.ClusterTopology{
+func MergeClusterTopologies(baseTopology resolved.ClusterTopology, clusterTopologies []resolved.ClusterTopology) (mergedTopology *resolved.ClusterTopology) {
+	mergedTopology = &resolved.ClusterTopology{
 		FlowID:              "all",
 		Services:            deepCopySlice(baseTopology.Services),
 		ServiceDependencies: deepCopySlice(baseTopology.ServiceDependencies),
@@ -19,11 +20,13 @@ func MergeClusterTopologies(baseTopology resolved.ClusterTopology, clusterTopolo
 		mergedTopology.Ingresses = append(mergedTopology.Ingresses, topology.Ingresses...)
 	}
 
-	mergedTopology.Services = lo.Uniq(mergedTopology.Services)
-	mergedTopology.ServiceDependencies = lo.Uniq(mergedTopology.ServiceDependencies)
+	//TODO improve the filtering method, we could implement the `Service.Equal` method to compare and filter the services
+	//TODO and inside this method we could use the k8s service marshall method (https://pkg.go.dev/k8s.io/api/core/v1#Service.Marsha) and also the same for other k8s fields
+	//TODO it should be faster
+	mergedTopology.Services = lo.UniqBy(mergedTopology.Services, MustGetMarshalledKey[*resolved.Service])
+	mergedTopology.ServiceDependencies = lo.UniqBy(mergedTopology.ServiceDependencies, MustGetMarshalledKey[resolved.ServiceDependency])
 	mergedTopology.Ingresses = foldAllIngress(mergedTopology.Ingresses)
 
-	// fmt.Printf("topology: %s\n", SPrintJSONClusterTopology(mergedTopology))
 	return mergedTopology
 }
 
@@ -39,3 +42,11 @@ func foldAllIngress(ingresses []*resolved.Ingress) []*resolved.Ingress {
 		return &merged
 	})
 }
+
+func MustGetMarshalledKey[T any](resource T) string {
+	bytes, err := json.Marshal(resource)
+	if err != nil {
+		panic("Failed to marshal resource")
+	}
+	return string(bytes)
+}

kontrol-service/main.go

@@ -2,9 +2,11 @@ package main
 
 import (
 	"flag"
+	"net/http"
 	"os"
 	"strconv"
 
+	cli_api "github.com/kurtosis-tech/kardinal/libs/cli-kontrol-api/api/golang/server"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
 	"github.com/sirupsen/logrus"
@@ -13,16 +15,23 @@ import (
 )
 
 func main() {
-	devMode := flag.Bool("dev-mode", false, "Allow to run the service in local mode.")
+	var devMode bool
+	devMode = *flag.Bool("dev-mode", false, "Allow to run the service in local mode.")
+	if !devMode {
+		devModeEnvVarStr := os.Getenv("DEV_MODE")
+		if devModeEnvVarStr == "true" {
+			devMode = true
+		}
+	}
 
 	flag.Parse()
 
-	if *devMode {
+	if devMode {
 		logrus.Warn("Running in dev mode. CORS fully open.")
 		logrus.SetLevel(logrus.DebugLevel)
 	}
 
-	startServer(*devMode)
+	startServer(devMode)
 }
 
 func startServer(isDevMode bool) {
@@ -62,7 +71,8 @@ func startServer(isDevMode bool) {
 
 	// Create a new Segment analytics client instance.
 	// analyticsClient is not initialized in dev mode so events are not reported to Segment
-	analyticsWrapper := api.NewAnalyticsWrapper(isDevMode)
+	analyticsWriteKeyEnvVarStr := os.Getenv("ANALYTICS_WRITE_KEY")
+	analyticsWrapper := api.NewAnalyticsWrapper(isDevMode, analyticsWriteKeyEnvVarStr)
 	defer analyticsWrapper.Close()
 
 	// create a type that satisfies the `api.ServerInterface`, which contains an implementation of every operation from the generated code
@@ -87,6 +97,34 @@ func startServer(isDevMode bool) {
 		},
 	}))
 
+	// Panic handler
+	e.Use(func(next echo.HandlerFunc) echo.HandlerFunc {
+		return func(c echo.Context) error {
+			defer func() {
+				if r := recover(); r != nil {
+
+					internalServerErrorResponse := cli_api.ErrorJSONResponse{
+						Error: "internal server error",
+					}
+
+					// Handle the panic and return a 500 error response
+					c.JSON(http.StatusInternalServerError, internalServerErrorResponse)
+					var debugMsg string
+					switch recoverErr := r.(type) {
+					case string:
+						debugMsg = recoverErr
+					case error:
+						debugMsg = recoverErr.Error()
+					default:
+						debugMsg = "recover didn't get error msg"
+					}
+					logrus.Errorf("HTTP server handle this internal panic: %s", debugMsg)
+				}
+			}()
+			return next(c)
+		}
+	})
+
 	server.RegisterExternalAndInternalApi(e)
 
 	// And we serve HTTP until the world ends.