[OSCP]feat:add kuscia task Related observational indicators

cmd/kuscia/modules/ktexporter.go

@@ -0,0 +1,56 @@
+// Copyright 2023 Ant Group Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package modules
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	pkgcom "github.com/secretflow/kuscia/pkg/common"
+	"github.com/secretflow/kuscia/pkg/ktexporter"
+	"github.com/secretflow/kuscia/pkg/utils/readyz"
+)
+
+type ktExporterModule struct {
+	moduleRuntimeBase
+	runMode            pkgcom.RunModeType
+	domainID           string
+	rootDir            string
+	metricUpdatePeriod uint
+	ktExportPort       string
+}
+
+func NewKtExporter(i *ModuleRuntimeConfigs) (Module, error) {
+	readyURI := fmt.Sprintf("http://127.0.0.1:%s", i.KtExportPort)
+	return &ktExporterModule{
+		moduleRuntimeBase: moduleRuntimeBase{
+			name:         "ktexporter",
+			readyTimeout: 60 * time.Second,
+			rdz: readyz.NewHTTPReadyZ(readyURI, 404, func(body []byte) error {
+				return nil
+			}),
+		},
+		runMode:            i.RunMode,
+		domainID:           i.DomainID,
+		rootDir:            i.RootDir,
+		metricUpdatePeriod: i.MetricUpdatePeriod,
+		ktExportPort:       i.KtExportPort,
+	}, nil
+}
+
+func (exporter *ktExporterModule) Run(ctx context.Context) error {
+	return ktexporter.KtExporter(ctx, exporter.runMode, exporter.domainID, exporter.metricUpdatePeriod, exporter.ktExportPort)
+}

cmd/kuscia/modules/metricexporter.go

@@ -51,6 +51,7 @@ func NewMetricExporter(i *ModuleRuntimeConfigs) (Module, error) {
 			"node-exporter": "http://localhost:" + i.NodeExportPort + "/metrics",
 			"envoy":         envoyexporter.GetEnvoyMetricURL(),
 			"ss":            "http://localhost:" + i.SsExportPort + "/ssmetrics",
+			"kt":            "http://localhost:" + i.KtExportPort + "/ktmetrics",
 		},
 	}, nil
 }

cmd/kuscia/modules/runtime.go

@@ -51,6 +51,7 @@ type ModuleRuntimeConfigs struct {
 	TransportPort           int
 	InterConnSchedulerPort  int
 	SsExportPort            string
+	KtExportPort            string
 	NodeExportPort          string
 	MetricExportPort        string
 	KusciaKubeConfig        string
@@ -190,6 +191,7 @@ func NewModuleRuntimeConfigs(ctx context.Context, kusciaConf confloader.KusciaCo
 	}
 
 	// init exporter ports
+	dependencies.KtExportPort = "9093"
 	dependencies.SsExportPort = "9092"
 	dependencies.NodeExportPort = "9100"
 	dependencies.MetricExportPort = "9091"

cmd/kuscia/start/start.go

@@ -90,6 +90,7 @@
 	mm.Regist("metricexporter", modules.NewMetricExporter, autonomy, lite, master)
 	mm.Regist("nodeexporter", modules.NewNodeExporter, autonomy, lite, master)
 	mm.Regist("ssexporter", modules.NewSsExporter, autonomy, lite, master)
+	mm.Regist("ktexporter", modules.NewKtExporter, autonomy, lite, master)
 	mm.Regist("scheduler", modules.NewScheduler, autonomy, master)
 	mm.Regist("transport", modules.NewTransport, autonomy, lite)
 
@@ -104,7 +105,8 @@
 	mm.SetDependencies("kusciaapi", "k3s", "config", "domainroute")
 	mm.SetDependencies("scheduler", "k3s")
 	mm.SetDependencies("ssexporter", "envoy")
-	mm.SetDependencies("metricexporter", "envoy", "ssexporter", "nodeexporter")
+	mm.SetDependencies("ktexporter", "envoy")
+	mm.SetDependencies("metricexporter", "envoy", "ssexporter", "ktexporter", "nodeexporter")
 	mm.SetDependencies("transport", "envoy")
 
 	mm.AddReadyHook(func(ctx context.Context, mdls map[string]modules.Module) error {

go.mod

@@ -271,6 +271,7 @@ require (
 )
 
 replace (
+	github.com/secretflow/kuscia/pkg/ktexporter => ./pkg/ktexporter
 	k8s.io/api => k8s.io/api v0.26.11
 	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.26.11
 	k8s.io/apimachinery => k8s.io/apimachinery v0.26.11

pkg/ktexporter/ktexporter.go

@@ -0,0 +1,88 @@
+// Copyright 2023 Ant Group Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ktexporter
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"github.com/secretflow/kuscia/pkg/ktexporter/ktmetrics"
+	"github.com/secretflow/kuscia/pkg/ktexporter/ktpromexporter"
+	"github.com/secretflow/kuscia/pkg/ktexporter/parse"
+
+	pkgcom "github.com/secretflow/kuscia/pkg/common"
+	"github.com/secretflow/kuscia/pkg/utils/nlog"
+)
+
+var (
+	ReadyChan = make(chan struct{})
+)
+
+func KtExporter(ctx context.Context, runMode pkgcom.RunModeType, domainID string, exportPeriod uint, port string) error {
+	// read the config
+	_, AggregationMetrics := parse.LoadMetricConfig()
+	clusterAddresses, _ := parse.GetClusterAddress(domainID)
+	localDomainName := domainID
+	var MetricTypes = ktpromexporter.NewMetricTypes_kt()
+
+	reg := ktpromexporter.ProduceRegister()
+	lastClusterMetricValues, err := ktmetrics.GetKtMetricResults(runMode, localDomainName, clusterAddresses, AggregationMetrics, exportPeriod)
+	if err != nil {
+		nlog.Warnf("Fail to get kt metric results, err: %v", err)
+		return err
+	}
+	// export the cluster metrics
+	ticker := time.NewTicker(time.Duration(exportPeriod) * time.Second)
+	defer ticker.Stop()
+	go func(runMode pkgcom.RunModeType, reg *prometheus.Registry, MetricTypes map[string]string, exportPeriods uint, lastClusterMetricValues map[string]float64) {
+		for range ticker.C {
+			// get clusterName and clusterAddress
+			clusterAddresses, _ = parse.GetClusterAddress(domainID)
+			// get cluster metrics
+			currentClusterMetricValues, err := ktmetrics.GetKtMetricResults(runMode, localDomainName, clusterAddresses, AggregationMetrics, exportPeriods)
+			if err != nil {
+				nlog.Warnf("Fail to get kt metric results, err: %v", err)
+			}
+
+			// calculate the change values of cluster metrics
+
+			ktpromexporter.UpdateMetrics(reg, currentClusterMetricValues, MetricTypes)
+		}
+	}(runMode, reg, MetricTypes, exportPeriod, lastClusterMetricValues)
+	// export to the prometheus
+	ktServer := http.NewServeMux()
+	ktServer.Handle("/ktmetrics", promhttp.HandlerFor(
+		reg,
+		promhttp.HandlerOpts{
+			EnableOpenMetrics: true,
+		}))
+	go func() {
+		if err := http.ListenAndServe("0.0.0.0:"+port, ktServer); err != nil {
+			nlog.Error("Fail to start the metric exporterserver", err)
+		}
+	}()
+	defer func() {
+		close(ReadyChan)
+		nlog.Info("Start to export metrics...")
+	}()
+
+	<-ctx.Done()
+	nlog.Info("Stopping the metric exporter...")
+	return nil
+}