From b7ce40d1d97711b50cb4983c5e5d0bce213898bc Mon Sep 17 00:00:00 2001
From: ddl <ddl196526@163.com>
Date: Wed, 31 Jul 2024 15:41:35 +0800
Subject: [PATCH] save

---
 cmd/run.go | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 52 insertions(+), 1 deletion(-)

diff --git a/cmd/run.go b/cmd/run.go
index e05b8ac..e9de03f 100644
--- a/cmd/run.go
+++ b/cmd/run.go
@@ -7,6 +7,8 @@ import (
 	"log/slog"
 	"net/http"
 	"os"
+	"os/exec"
+	"path/filepath"
 
 	"github.com/julienschmidt/httprouter"
 	"github.com/pkg/errors"
@@ -24,6 +26,11 @@ import (
 	eigensdktypes "github.com/Layr-Labs/eigensdk-go/types"
 )
 
+const (
+	FlinkVersion = "1.18.1"
+	FlinkDir     = "/opt/flink"
+)
+
 var runCmd = &cobra.Command{
 	Use:   "run",
 	Short: "monitor manualscript requests and send signed to gateway",
@@ -37,7 +44,7 @@ var runCmd = &cobra.Command{
 
 		router := httprouter.New()
 		router.GET("/eigen/node/health", handleHealth)
-		
+
 		err = http.ListenAndServe(fmt.Sprintf(":%d", HealthCheckPort), router)
 		if err != nil {
 			log.Println(err)
@@ -104,5 +111,49 @@ func Run(ctx context.Context) error {
 		}
 
 	}
+
+	// check if flink is installed
+	if _, err := os.Stat(FlinkDir); os.IsNotExist(err) {
+		// 安装 Flink
+		if err := installFlink(); err != nil {
+			return fmt.Errorf("failed to install Flink: %w", err)
+		}
+	}
+
+	//
+	if err := startFlink(); err != nil {
+		return fmt.Errorf("failed to start Flink: %w", err)
+	}
+
 	return errors.New("AVS is not registered")
 }
+
+func installFlink() error {
+	// 下载 Flink
+	cmd := exec.Command("wget", fmt.Sprintf("https://dlcdn.apache.org/flink/flink-%s/flink-%s-bin-scala_2.12.tgz", FlinkVersion, FlinkVersion))
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to download Flink: %w", err)
+	}
+
+	// 解压 Flink
+	cmd = exec.Command("tar", "xzf", fmt.Sprintf("flink-%s-bin-scala_2.12.tgz", FlinkVersion))
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to extract Flink: %w", err)
+	}
+
+	// 移动 Flink 到 /opt 目录
+	cmd = exec.Command("sudo", "mv", fmt.Sprintf("flink-%s", FlinkVersion), FlinkDir)
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to move Flink to %s: %w", FlinkDir, err)
+	}
+
+	return nil
+}
+
+func startFlink() error {
+	cmd := exec.Command(filepath.Join(FlinkDir, "bin", "start-cluster.sh"))
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to start Flink cluster: %w", err)
+	}
+	return nil
+}