Merge pull request #201 from swanchain/fea-sync-mining-status

Add sync mining task status and show `rejected` tasks

Author: Normalnoise

README.md

@@ -298,7 +298,7 @@ spec:
     spec:
       containers:
       - name: resource-exporter
-        image: filswan/resource-exporter:v11.3.1
+        image: filswan/resource-exporter:v11.3.2
         imagePullPolicy: IfNotPresent
         securityContext:
           privileged: true
@@ -355,6 +355,7 @@ make install
        WalletBlackList = ""                           # CP reject user addresses from this blacklist for space deployment
        Pricing = "true"                               # default True, indicating acceptance of smart pricing orders, which may include orders priced lower than self-determined pricing.
        AutoDeleteImage = false                        # Default false, automatically delete unused images
+       ClearLogDuration = 24                          # The interval for automatically clearing the log, the unit is hours
        PortRange= ["40000-40050","40070"]             # Externally exposed port number for deploying ECP image tasks
       
        [UBI]

build/version.go

@@ -10,13 +10,13 @@ var CurrentCommit string
 
 var NetWorkTag string
 
-const BuildVersion = "1.0.1"
+const BuildVersion = "1.0.2"
 
 const UBITaskImageIntelCpu = "filswan/ubi-worker-cpu-intel:latest"
 const UBITaskImageIntelGpu = "filswan/ubi-worker-gpu-intel:latest"
 const UBITaskImageAmdCpu = "filswan/ubi-worker-cpu-amd:latest"
 const UBITaskImageAmdGpu = "filswan/ubi-worker-gpu-amd:latest"
-const UBIResourceExporterDockerImage = "filswan/resource-exporter:v11.3.1"
+const UBIResourceExporterDockerImage = "filswan/resource-exporter:v11.3.2"
 const TraefikServerDockerImage = "traefik:v2.10"
 
 func UserVersion() string {

cmd/computing-provider/ubi.go

@@ -239,6 +239,8 @@ var daemonCmd = &cli.Command{
 func getStatusColor(taskStatus int) []tablewriter.Colors {
 	var rowColor []tablewriter.Colors
 	switch taskStatus {
+	case models.TASK_REJECTED_STATUS:
+		rowColor = []tablewriter.Colors{{tablewriter.Bold, tablewriter.FgRedColor}}
 	case models.TASK_RECEIVED_STATUS:
 		rowColor = []tablewriter.Colors{{tablewriter.Bold, tablewriter.FgYellowColor}}
 	case models.TASK_RUNNING_STATUS:

conf/config.go

@@ -46,15 +46,16 @@ type ComputeNode struct {
 }
 
 type API struct {
-	Port            int
-	MultiAddress    string
-	Domain          string
-	NodeName        string
-	WalletWhiteList string
-	WalletBlackList string
-	Pricing         Pricing  `toml:"pricing"`
-	AutoDeleteImage bool     `toml:"AutoDeleteImage"`
-	PortRange       []string `toml:"PortRange"`
+	Port             int
+	MultiAddress     string
+	Domain           string
+	NodeName         string
+	WalletWhiteList  string
+	WalletBlackList  string
+	Pricing          Pricing  `toml:"pricing"`
+	AutoDeleteImage  bool     `toml:"AutoDeleteImage"`
+	ClearLogDuration int      `toml:"ClearLogDuration"`
+	PortRange        []string `toml:"PortRange"`
 }
 type UBI struct {
 	UbiEnginePk     string
@@ -71,7 +72,6 @@ type LOG struct {
 }
 
 type HUB struct {
-	AccessToken      string
 	BalanceThreshold float64
 	OrchestratorPk   string
 	VerifySign       bool
@@ -216,6 +216,9 @@ func getConfigByHeight() {
 			config.UBI.EdgeUrl = ncCopy.Config.EdgeUrl
 			config.CONTRACT.ZkCollateralUbiZero = ncCopy.Config.ZkCollateralUbiZeroContract
 			config.CONTRACT.JobCollateralUbiZero = ncCopy.Config.OrchestratorCollateralUbiZeroContract
+			if config.API.ClearLogDuration == 0 {
+				config.API.ClearLogDuration = 24
+			}
 		}
 	}
 }

config.toml.sample

@@ -7,6 +7,7 @@ WalletWhiteList = ""                                                     # CP ac
 WalletBlackList = ""                                                     # CP reject user addresses from this blacklist for space deployment
 Pricing = "true"                                                         # Default True, indicating acceptance of smart pricing orders, which may include orders priced lower than self-determined pricing.
 AutoDeleteImage = false                                                  # Default false, automatically delete unused images
+ClearLogDuration = 24                                                    # Delete logs at intervals after the job is finished, the unit is hours
 PortRange = ["40000-40050","40070"]                                      # Externally exposed port number for deploying multi-port image tasks
 
 

constants/utils.go

@@ -1,16 +1,5 @@
 package constants
 
-const StatusActive = "Active"
-const StatusOffline = "Offline"
-
-// bidding status
-const BiddingCreated string = "created"
-const BiddingAccepting string = "accepting_bids"
-const BiddingProcessing string = "processing"
-const BiddingSubmitted string = "submitted"
-const BiddingCompleted string = "completed"
-const BiddingCancelled string = "cancelled"
-
 const K8S_NAMESPACE_NAME_PREFIX = "ns-"
 const K8S_CONTAINER_NAME_PREFIX = "pod-"
 const K8S_PRIVATE_CONTAINER_PREFIX = "vm-"
@@ -29,3 +18,10 @@ const (
 	BidMode_Private
 	BidMode_None
 )
+
+const (
+	BUILD_IMAGE_PATH_PREFIX = "build"
+	LOG_PATH_PREFIX         = "logs"
+	BUILD_LOG_NAME          = "build.log"
+	Container_LOG_NAME      = "container.log"
+)

internal/computing/buildspace.go

@@ -122,7 +122,7 @@ func BuildImagesByDockerfile(jobUuid, spaceName, imagePath string) (string, stri
 	log.Printf("Image path: %s", imagePath)
 
 	dockerService := NewDockerService()
-	if err := dockerService.BuildImage(imagePath, imageName); err != nil {
+	if err := dockerService.BuildImage(jobUuid, imagePath, imageName); err != nil {
 		logs.GetLogger().Errorf("Error building Docker image: %v", err)
 		return "", ""
 	}

internal/computing/cron_task.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"github.com/swanchain/go-computing-provider/internal/contract"
+	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -45,6 +47,9 @@ func (task *CronTask) RunTask() {
 	task.checkJobReward()
 	task.cleanImageResource()
 	task.CheckCpBalance()
+	task.UpdateContainerLog()
+	task.DeleteSpaceLog()
+
 }
 
 func CheckClusterNetworkPolicy() {
@@ -566,7 +571,7 @@ func addNodeLabel() {
 
 func (task *CronTask) CheckCpBalance() {
 	c := cron.New(cron.WithSeconds())
-	c.AddFunc("* 0/30 * * * ?", func() {
+	c.AddFunc("0 0/30 * * * ?", func() {
 		defer func() {
 			if err := recover(); err != nil {
 				logs.GetLogger().Errorf("check cp balance catch panic error: %+v", err)
@@ -577,6 +582,60 @@ func (task *CronTask) CheckCpBalance() {
 	c.Start()
 }
 
+func (task *CronTask) UpdateContainerLog() {
+	c := cron.New(cron.WithSeconds())
+	c.AddFunc("* 0/10 * * * ?", func() {
+		defer func() {
+			if err := recover(); err != nil {
+				logs.GetLogger().Errorf("update container log catch panic error: %+v", err)
+			}
+		}()
+
+		jobList, err := NewJobService().GetJobList(models.UN_DELETEED_FLAG, -1)
+		if err != nil {
+			logs.GetLogger().Errorf("failed to get job data, error: %+v", err)
+			return
+		}
+
+		for _, job := range jobList {
+			NewK8sService().UpdateContainerLogToFile(job.JobUuid)
+		}
+	})
+	c.Start()
+}
+
+func (task *CronTask) DeleteSpaceLog() {
+	c := cron.New(cron.WithSeconds())
+	c.AddFunc("0 0/30 * * * ?", func() {
+		defer func() {
+			if err := recover(); err != nil {
+				logs.GetLogger().Errorf("update container log catch panic error: %+v", err)
+			}
+		}()
+
+		jobList, err := NewJobService().GetJobList(models.DELETED_FLAG, -1)
+		if err != nil {
+			logs.GetLogger().Errorf("failed to get job data, error: %+v", err)
+			return
+		}
+		cpRepoPath, _ := os.LookupEnv("CP_PATH")
+
+		for _, job := range jobList {
+			if job.CreateTime+int64(24*7*3600) < time.Now().Unix() {
+				continue
+			}
+			if job.ExpireTime+int64(conf.GetConfig().API.ClearLogDuration)*3600 < time.Now().Unix() {
+				err := os.RemoveAll(filepath.Join(cpRepoPath, constants.LOG_PATH_PREFIX, job.JobUuid))
+				if err != nil {
+					logs.GetLogger().Errorf("failed to delete logs, job_uuid: %s, error: %v", job.JobUuid, err)
+					continue
+				}
+			}
+		}
+	})
+	c.Start()
+}
+
 func reportJobStatus(jobUuid string, deployStatus int) bool {
 	var job = new(models.JobEntity)
 	job.JobUuid = jobUuid
@@ -659,7 +718,8 @@ func checkFcpJobInfoInChain(job *models.JobEntity) {
 type TaskGroup struct {
 	Items []*models.TaskEntity
 	Ids   []int64
-	Type  int // 1: contract  2: sequncer
+	Uuids []string
+	Type  int // 1: contract  2: sequncer 3: mining
 }
 
 func handleTasksToGroup(list []*models.TaskEntity) []TaskGroup {
@@ -695,3 +755,23 @@ func handleTasksToGroup(list []*models.TaskEntity) []TaskGroup {
 	}
 	return groups
 }
+
+func handleTasksToGroupForMining(list []*models.TaskEntity) []TaskGroup {
+	var groups []TaskGroup
+	var group TaskGroup
+
+	const batchSize = 10
+	for i := 0; i < len(list); i++ {
+		if len(group.Items) > batchSize {
+			groups = append(groups, group)
+			group = TaskGroup{}
+		}
+		group.Items = append(group.Items, list[i])
+		group.Uuids = append(group.Uuids, list[i].Uuid)
+		group.Type = 3
+	}
+	if len(group.Items) > 0 {
+		groups = append(groups, group)
+	}
+	return groups
+}

internal/computing/deploy.go

@@ -419,7 +419,10 @@ func (d *Deploy) ModelInferenceToK8s() error {
 
 	imageName := "lagrange/" + modelInfo.Framework + ":v1.0"
 
-	logFile := filepath.Join(d.SpacePath, BuildFileName)
+	logDir := filepath.Join(cpPath, constants.LOG_PATH_PREFIX, d.originalJobUuid)
+	os.MkdirAll(logDir, os.ModePerm)
+
+	logFile := filepath.Join(logDir, constants.BUILD_LOG_NAME)
 	if _, err = os.Create(logFile); err != nil {
 		return err
 	}

internal/computing/docker_service.go

@@ -17,6 +17,7 @@ import (
 	"github.com/docker/docker/api/types/registry"
 	"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"
 	"os/exec"
@@ -32,8 +33,6 @@ import (
 	"github.com/docker/docker/client"
 )
 
-const BuildFileName = "build.log"
-
 type DockerService struct {
 	c *client.Client
 }
@@ -75,7 +74,7 @@ func ExtractExposedPort(dockerfilePath string) (string, error) {
 	return exposedPort, nil
 }
 
-func (ds *DockerService) BuildImage(buildPath, imageName string) error {
+func (ds *DockerService) BuildImage(jobUuid, buildPath, imageName string) error {
 	// Create a buffer
 	buf := new(bytes.Buffer)
 	tw := tar.NewWriter(buf)
@@ -120,7 +119,11 @@ func (ds *DockerService) BuildImage(buildPath, imageName string) error {
 	}
 	defer buildResponse.Body.Close()
 
-	logFile, err := os.Create(filepath.Join(buildPath, BuildFileName))
+	cpRepoPath, _ := os.LookupEnv("CP_PATH")
+	logDir := filepath.Join(cpRepoPath, constants.LOG_PATH_PREFIX, jobUuid)
+	os.MkdirAll(logDir, os.ModePerm)
+
+	logFile, err := os.Create(filepath.Join(logDir, constants.BUILD_LOG_NAME))
 	if err != nil {
 		return err
 	}