Merge pull request #235 from swanchain/feature/expose-metrics-endpoints

Expose metrics endpoints

Author: swan-alexchen

build/version.go

@@ -16,10 +16,10 @@ const UBITaskImageIntelCpu = "swanhub/ubi-worker-cpu-intel:latest"
 const UBITaskImageIntelGpu = "swanhub/ubi-worker-gpu-intel:latest"
 const UBITaskImageAmdCpu = "swanhub/ubi-worker-cpu-amd:latest"
 const UBITaskImageAmdGpu = "swanhub/ubi-worker-gpu-amd:latest"
-const UBIResourceExporterDockerImage = "swanhub/resource-exporter:v12.0.0"
+const UBIResourceExporterDockerImage = "swanhub/resource-exporter:v13.0.0"
 const TraefikServerDockerImage = "traefik:v2.10"
 
-const ResourceExporterVersion = "v12.0.0"
+const ResourceExporterVersion = "v13.0.0"
 
 func UserVersion() string {
 	return BuildVersion + "+" + NetWorkTag + CurrentCommit

cmd/computing-provider/run.go

@@ -3,11 +3,18 @@ package main
 import (
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/filswan/go-mcs-sdk/mcs/api/common/logs"
 	"github.com/gin-contrib/pprof"
 	"github.com/gin-gonic/gin"
-	"github.com/itsjamie/gin-cors"
+	cors "github.com/itsjamie/gin-cors"
 	"github.com/olekukonko/tablewriter"
 	"github.com/swanchain/go-computing-provider/build"
 	"github.com/swanchain/go-computing-provider/conf"
@@ -21,12 +28,6 @@ import (
 	"github.com/swanchain/go-computing-provider/util"
 	"github.com/swanchain/go-computing-provider/wallet"
 	"github.com/urfave/cli/v2"
-	"os"
-	"path/filepath"
-	"regexp"
-	"strconv"
-	"strings"
-	"time"
 )
 
 var runCmd = &cli.Command{
@@ -76,6 +77,7 @@ var runCmd = &cli.Command{
 func cpManager(router *gin.RouterGroup) {
 	router.GET("/cp", computing.StatisticalSources)
 	router.GET("/host/info", computing.GetServiceProviderInfo)
+	router.GET("/cp/metrics", computing.GetResourceExporterMetrics)
 	router.POST("/lagrange/jobs", computing.ReceiveJob)
 	router.DELETE("/lagrange/jobs", computing.CancelJob)
 	router.POST("/lagrange/jobs/renew", computing.ReNewJob)

cmd/computing-provider/ubi.go

@@ -3,6 +3,11 @@ package main
 import (
 	_ "embed"
 	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"time"
+
 	"github.com/filswan/go-mcs-sdk/mcs/api/common/logs"
 	"github.com/gin-contrib/pprof"
 	"github.com/gin-gonic/gin"
@@ -14,10 +19,6 @@ import (
 	"github.com/swanchain/go-computing-provider/internal/models"
 	"github.com/swanchain/go-computing-provider/util"
 	"github.com/urfave/cli/v2"
-	"os"
-	"path/filepath"
-	"strconv"
-	"time"
 )
 
 var ubiTaskCmd = &cli.Command{
@@ -219,6 +220,7 @@ var daemonCmd = &cli.Command{
 
 		router := r.Group("/api/v1/computing")
 		router.GET("/cp", computing.GetCpResource)
+		router.GET("/cp/metrics", computing.GetUbiResourceExporterMetrics)
 		router.POST("/cp/ubi", computing.DoUbiTaskForDocker)
 		router.POST("/cp/docker/receive/ubi", computing.ReceiveUbiProof)
 

internal/computing/deploy.go

@@ -527,7 +527,7 @@ func (d *Deploy) ModelInferenceToK8s() error {
 func (d *Deploy) configureSshAccess(k8sService *K8sService, podName string) error {
 
 	sshkeyCmd := []string{"sh", "-c", fmt.Sprintf("echo '%s' > /root/.ssh/authorized_keys", d.sshKey)}
-	if err := k8sService.PodDoCommand(d.k8sNameSpace, podName, "", sshkeyCmd); err != nil {
+	if _, _, err := k8sService.PodDoCommand(d.k8sNameSpace, podName, "", sshkeyCmd); err != nil {
 		return fmt.Errorf("failed to add sshkey, job_uuid: %s error: %v", d.jobUuid, err)
 	}
 

internal/computing/docker_service.go

@@ -9,6 +9,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"time"
+
 	"github.com/containerd/containerd/namespaces"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
@@ -18,12 +25,6 @@ import (
 	"github.com/filswan/go-mcs-sdk/mcs/api/common/logs"
 	"github.com/swanchain/go-computing-provider/build"
 	"github.com/swanchain/go-computing-provider/constants"
-	"io"
-	"os"
-	"path/filepath"
-	"regexp"
-	"strings"
-	"time"
 
 	"github.com/swanchain/go-computing-provider/conf"
 
@@ -396,6 +397,41 @@ func (ds *DockerService) ContainerLogs(containerName string) (string, error) {
 	}
 }
 
+func (ds *DockerService) ContainerExec(containerID string, cmd []string) (string, error) {
+	ctx := context.Background()
+	execConfig := container.ExecOptions{
+		AttachStdout: true,
+		AttachStderr: true,
+		Cmd:          cmd,
+	}
+	execIDResponse, err := ds.c.ContainerExecCreate(ctx, containerID, execConfig)
+	if err != nil {
+		return "", fmt.Errorf("failed to create exec command: %w", err)
+	}
+
+	if execIDResponse.ID == "" {
+		return "", fmt.Errorf("exec ID is empty")
+	}
+
+	resp, err := ds.c.ContainerExecAttach(ctx, execIDResponse.ID, container.ExecAttachOptions{})
+	if err != nil {
+		return "", fmt.Errorf("failed to attach to exec command: %w", err)
+	}
+	defer resp.Close()
+
+	var outBuf, errBuf bytes.Buffer
+	_, err = io.Copy(&outBuf, resp.Reader)
+	if err != nil {
+		return "", fmt.Errorf("failed to read exec output: %w", err)
+	}
+
+	if errBuf.Len() > 0 {
+		return "", fmt.Errorf("exec command stderr: %s", errBuf.String())
+	}
+
+	return outBuf.String(), nil
+}
+
 func (ds *DockerService) GetContainerLogStream(ctx context.Context, containerName string) (io.ReadCloser, error) {
 	return ds.c.ContainerLogs(ctx, containerName, container.LogsOptions{
 		ShowStdout: true,

internal/computing/k8s_service.go

@@ -2,6 +2,7 @@ package computing
 
 import "C"
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -556,6 +557,7 @@ func (s *K8sService) GetResourceExporterVersion() (string, error) {
 	}
 	return version, nil
 }
+
 func (s *K8sService) GetPodLogByPodName(namespace, podName string, podLogOptions *coreV1.PodLogOptions) (string, error) {
 	req := s.k8sClient.CoreV1().Pods(namespace).GetLogs(podName, podLogOptions)
 	buf, err := readLog(req)
@@ -675,8 +677,8 @@ func (s *K8sService) WaitForPodRunningByTcp(namespace, jobUuid string, labelSele
 	return podName, nil
 }
 
-func (s *K8sService) PodDoCommand(namespace, podName, containerName string, podCmd []string) error {
-	reader, writer := io.Pipe()
+func (s *K8sService) PodDoCommand(namespace, podName, containerName string, podCmd []string) (string, string, error) {
+	var stdout, stderr bytes.Buffer
 	req := s.k8sClient.CoreV1().RESTClient().
 		Post().
 		Resource("pods").
@@ -686,28 +688,30 @@ func (s *K8sService) PodDoCommand(namespace, podName, containerName string, podC
 		VersionedParams(&coreV1.PodExecOptions{
 			Container: containerName,
 			Command:   podCmd,
-			Stdin:     true,
+			Stdin:     false,
 			Stdout:    true,
 			Stderr:    true,
-			TTY:       true,
+			TTY:       false,
 		}, scheme.ParameterCodec)
 
 	executor, err := remotecommand.NewSPDYExecutor(s.config, "POST", req.URL())
 	if err != nil {
-		return fmt.Errorf("failed to create spdy client: %w", err)
+		logs.GetLogger().Errorf("Failed to create spdy client for pod %s: %v", podName, err)
+		return "", "", fmt.Errorf("failed to create spdy client: %w", err)
 	}
 
-	err = executor.Stream(remotecommand.StreamOptions{
-		Stdin:  reader,
-		Stdout: writer,
-		Stderr: writer,
-		Tty:    true,
+	err = executor.StreamWithContext(context.TODO(), remotecommand.StreamOptions{
+		Stdin:  nil,
+		Stdout: &stdout,
+		Stderr: &stderr,
+		Tty:    false,
 	})
 	if err != nil {
-		return fmt.Errorf("failed to create stream: %w", err)
+		logs.GetLogger().Errorf("Failed to create stream for pod %s: %v", podName, err)
+		return "", "", fmt.Errorf("failed to create stream: %w", err)
 	}
 
-	return nil
+	return stdout.String(), stderr.String(), nil
 }
 
 type GpuData struct {

internal/computing/space_service.go

@@ -2,6 +2,7 @@ package computing
 
 import (
 	"bufio"
+	"bytes"
 	"context"
 	"encoding/base64"
 	"encoding/json"
@@ -295,6 +296,75 @@ func ReceiveJob(c *gin.Context) {
 	c.JSON(http.StatusOK, util.CreateSuccessResponse(jobData))
 }
 
+func GetResourceExporterMetrics(c *gin.Context) {
+	logs.GetLogger().Info("Starting GetResourceExporterMetrics function.")
+	k8sService := NewK8sService()
+	if k8sService == nil {
+		logs.GetLogger().Info("Failed to create k8s service client.")
+		c.JSON(http.StatusInternalServerError, util.CreateErrorResponse(util.ServerError, "failed to create k8s service client"))
+		return
+	}
+	logs.GetLogger().Info("Successfully created k8s service client.")
+
+	logs.GetLogger().Info("Attempting to list resource-exporter pods in kube-system namespace.")
+	podList, err := k8sService.k8sClient.CoreV1().Pods("kube-system").List(context.Background(), metaV1.ListOptions{
+		LabelSelector: "app=resource-exporter",
+	})
+	if err != nil {
+		logs.GetLogger().Errorf("Failed to list resource-exporter pods, error: %v", err)
+		c.JSON(http.StatusInternalServerError, util.CreateErrorResponse(util.ServerError, fmt.Sprintf("failed to list resource-exporter pods: %v", err)))
+		return
+	}
+	logs.GetLogger().Infof("Successfully listed %d resource-exporter pods.", len(podList.Items))
+
+	if len(podList.Items) == 0 {
+		logs.GetLogger().Info("No resource-exporter pods found.")
+		c.JSON(http.StatusNotFound, util.CreateErrorResponse(util.ServerError, "resource-exporter pod not found"))
+		return
+	}
+	logs.GetLogger().Info("Resource-exporter pod found.")
+
+	// Assuming we only need metrics from one resource-exporter pod, pick the first one
+	resourceExporterPodName := podList.Items[0].Name
+	if resourceExporterPodName == "" {
+		logs.GetLogger().Info("Resource-exporter pod name is empty.")
+		c.JSON(http.StatusInternalServerError, util.CreateErrorResponse(util.ServerError, "resource-exporter pod name not found"))
+		return
+	}
+	logs.GetLogger().Infof("Resource-exporter pod name: %s", resourceExporterPodName)
+
+	var mergedMetrics bytes.Buffer
+
+	// Get /node/metrics using PodDoCommand
+	logs.GetLogger().Infof("Attempting to get /node/metrics from resource-exporter pod %s.", resourceExporterPodName)
+	// wget -q -O - localhost:9000/node/metrics
+	nodeMetricsCmd := []string{"wget", "-q", "-O", "-", "localhost:9000/node/metrics"}
+	nodeMetricsStdout, _, err := k8sService.PodDoCommand("kube-system", resourceExporterPodName, "", nodeMetricsCmd)
+	if err != nil {
+		logs.GetLogger().Errorf("Failed to get /node/metrics from resource-exporter pod %s: %v", resourceExporterPodName, err)
+		c.JSON(http.StatusInternalServerError, util.CreateErrorResponse(util.ServerError, fmt.Sprintf("failed to get node metrics: %v", err)))
+		return
+	}
+	logs.GetLogger().Info("Successfully retrieved /node/metrics.")
+	mergedMetrics.WriteString(nodeMetricsStdout)
+	mergedMetrics.WriteString("\n")
+
+	// Get /dcgm/metrics using PodDoCommand
+	logs.GetLogger().Infof("Attempting to get /dcgm/metrics from resource-exporter pod %s.", resourceExporterPodName)
+	dcgmMetricsCmd := []string{"wget", "-q", "-O", "-", "localhost:9000/dcgm/metrics"}
+	dcgmMetricsStdout, _, err := k8sService.PodDoCommand("kube-system", resourceExporterPodName, "", dcgmMetricsCmd)
+	if err != nil {
+		logs.GetLogger().Errorf("Failed to get /dcgm/metrics from resource-exporter pod %s: %v", resourceExporterPodName, err)
+		c.JSON(http.StatusInternalServerError, util.CreateErrorResponse(util.ServerError, fmt.Sprintf("failed to get dcgm metrics: %v", err)))
+		return
+	}
+	logs.GetLogger().Info("Successfully retrieved /dcgm/metrics.")
+	mergedMetrics.WriteString(dcgmMetricsStdout)
+
+	logs.GetLogger().Info("Returning merged resource-exporter metrics.")
+	c.String(http.StatusOK, mergedMetrics.String())
+}
+
 func submitJob(jobData *models.JobData) error {
 	cpRepoPath, ok := os.LookupEnv("CP_PATH")
 	if !ok {
@@ -1775,7 +1845,7 @@ func downloadModelUrl(namespace, jobUuid, serviceIp string, podCmd []string) {
 		return
 	}
 
-	if err = k8sService.PodDoCommand(namespace, podName, "", podCmd); err != nil {
+	if _, _, err = k8sService.PodDoCommand(namespace, podName, "", podCmd); err != nil {
 		logs.GetLogger().Error(err)
 		return
 	}