feat(publisher): implement fileserver publisher service (#187)

Signed-off-by: wuhuizuo <[email protected]> --------- Signed-off-by: wuhuizuo <[email protected]>
PingCAP-QE · Oct 30, 2024 · 30432bc · 30432bc
1 parent 2e1e85d
commit 30432bc
Show file tree

Hide file tree

Showing 32 changed files with 1,360 additions and 105 deletions.
diff --git a/publisher/cmd/publisher-cli/http.go b/publisher/cmd/publisher-cli/http.go
@@ -4,9 +4,10 @@ import (
 	"net/http"
 	"time"
 
-	cli "github.com/PingCAP-QE/ee-apps/publisher/gen/http/cli/publisher"
 	goahttp "goa.design/goa/v3/http"
 	goa "goa.design/goa/v3/pkg"
+
+	cli "github.com/PingCAP-QE/ee-apps/publisher/gen/http/cli/publisher"
 )
 
 func doHTTP(scheme, host string, timeout int, debug bool) (goa.Endpoint, any, error) {

diff --git a/publisher/cmd/publisher/http.go b/publisher/cmd/publisher/http.go
@@ -7,16 +7,19 @@ import (
 	"sync"
 	"time"
 
-	tiupsvr "github.com/PingCAP-QE/ee-apps/publisher/gen/http/tiup/server"
-	tiup "github.com/PingCAP-QE/ee-apps/publisher/gen/tiup"
 	"goa.design/clue/debug"
 	"goa.design/clue/log"
 	goahttp "goa.design/goa/v3/http"
+
+	"github.com/PingCAP-QE/ee-apps/publisher/gen/fileserver"
+	fssvr "github.com/PingCAP-QE/ee-apps/publisher/gen/http/fileserver/server"
+	tiupsvr "github.com/PingCAP-QE/ee-apps/publisher/gen/http/tiup/server"
+	tiup "github.com/PingCAP-QE/ee-apps/publisher/gen/tiup"
 )
 
 // handleHTTPServer starts configures and starts a HTTP server on the given
 // URL. It shuts down the server if any error is received in the error channel.
-func handleHTTPServer(ctx context.Context, u *url.URL, tiupEndpoints *tiup.Endpoints, wg *sync.WaitGroup, errc chan error, dbg bool) {
+func handleHTTPServer(ctx context.Context, u *url.URL, tiupEndpoints *tiup.Endpoints, fsEndpoints *fileserver.Endpoints, wg *sync.WaitGroup, errc chan error, dbg bool) {
 
 	// Provide the transport specific request decoder and response encoder.
 	// The goa http package has built-in support for JSON, XML and gob.
@@ -46,14 +49,17 @@ func handleHTTPServer(ctx context.Context, u *url.URL, tiupEndpoints *tiup.Endpo
 	// responses.
 	var (
 		tiupServer *tiupsvr.Server
+		fsServer   *fssvr.Server
 	)
 	{
 		eh := errorHandler(ctx)
 		tiupServer = tiupsvr.New(tiupEndpoints, mux, dec, enc, eh, nil)
+		fsServer = fssvr.New(fsEndpoints, mux, dec, enc, eh, nil)
 	}
 
 	// Configure the mux.
 	tiupsvr.Mount(mux, tiupServer)
+	fssvr.Mount(mux, fsServer)
 
 	var handler http.Handler = mux
 	if dbg {

diff --git a/publisher/cmd/publisher/main.go b/publisher/cmd/publisher/main.go
@@ -17,8 +17,9 @@ import (
 	"goa.design/clue/debug"
 	"goa.design/clue/log"
 
-	gentiup "github.com/PingCAP-QE/ee-apps/publisher/gen/tiup"
-	"github.com/PingCAP-QE/ee-apps/publisher/pkg/impl/tiup"
+	"github.com/PingCAP-QE/ee-apps/publisher/gen/fileserver"
+	"github.com/PingCAP-QE/ee-apps/publisher/gen/tiup"
+	"github.com/PingCAP-QE/ee-apps/publisher/pkg/impl"
 )
 
 func main() {
@@ -52,7 +53,7 @@ func main() {
 		logLevel = zerolog.DebugLevel
 	}
 	zerolog.SetGlobalLevel(logLevel)
-	logger := zerolog.New(os.Stderr).With().Timestamp().Str("service", gentiup.ServiceName).Logger()
+	logger := zerolog.New(os.Stderr).With().Timestamp().Str("service", tiup.ServiceName).Logger()
 
 	// Load and parse configuration
 	config, err := loadConfig(*configFile)
@@ -62,7 +63,8 @@ func main() {
 
 	// Initialize the services.
 	var (
-		tiupSvc gentiup.Service
+		tiupSvc tiup.Service
+		fsSvc   fileserver.Service
 	)
 	{
 		// Configure Kafka kafkaWriter
@@ -81,18 +83,23 @@ func main() {
 			DB:       config.Redis.DB,
 		})
 
-		tiupSvc = tiup.NewService(&logger, kafkaWriter, redisClient, config.EventSource)
+		tiupSvc = impl.NewTiup(&logger, kafkaWriter, redisClient, config.EventSource)
+		fsSvc = impl.NewFileserver(&logger, kafkaWriter, redisClient, config.EventSource)
 	}
 
 	// Wrap the services in endpoints that can be invoked from other services
 	// potentially running in different processes.
 	var (
-		tiupEndpoints *gentiup.Endpoints
+		tiupEndpoints *tiup.Endpoints
+		fsEndpoints   *fileserver.Endpoints
 	)
 	{
-		tiupEndpoints = gentiup.NewEndpoints(tiupSvc)
+		tiupEndpoints = tiup.NewEndpoints(tiupSvc)
 		tiupEndpoints.Use(debug.LogPayloads())
 		tiupEndpoints.Use(log.Endpoint)
+		fsEndpoints = fileserver.NewEndpoints(fsSvc)
+		fsEndpoints.Use(debug.LogPayloads())
+		fsEndpoints.Use(log.Endpoint)
 	}
 
 	// Create channel used by both the signal handler and server goroutines
@@ -134,7 +141,7 @@ func main() {
 			} else if u.Port() == "" {
 				u.Host = net.JoinHostPort(u.Host, "80")
 			}
-			handleHTTPServer(ctx, u, tiupEndpoints, &wg, errc, *dbgF)
+			handleHTTPServer(ctx, u, tiupEndpoints, fsEndpoints, &wg, errc, *dbgF)
 		}
 
 	default:

diff --git a/publisher/cmd/worker/main.go b/publisher/cmd/worker/main.go
@@ -19,7 +19,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	"github.com/PingCAP-QE/ee-apps/publisher/pkg/config"
-	"github.com/PingCAP-QE/ee-apps/publisher/pkg/impl/tiup"
+	"github.com/PingCAP-QE/ee-apps/publisher/pkg/impl"
 )
 
 func main() {
@@ -60,14 +60,14 @@ func main() {
 
 	ctx := log.Logger.WithContext(context.Background())
 	// Create TiUP publisher
-	var handler *tiup.Publisher
+	var handler *impl.Worker
 	{
 		var err error
 		nigthlyInterval, err := time.ParseDuration(config.Options.NightlyInterval)
 		if err != nil {
 			log.Fatal().Err(err).Msg("Error parsing nightly interval")
 		}
-		handler, err = tiup.NewPublisher(&log.Logger, redisClient, tiup.PublisherOptions{
+		handler, err = impl.NewWorker(&log.Logger, redisClient, impl.WorkerOptions{
 			MirrorURL:       config.Options.MirrorURL,
 			LarkWebhookURL:  config.Options.LarkWebhookURL,
 			NightlyInterval: nigthlyInterval,
@@ -106,7 +106,7 @@ func main() {
 	var wg sync.WaitGroup
 	ctx, cancel := context.WithCancel(ctx)
 
-	Start(ctx, reader, handler, &wg, errc)
+	start(ctx, reader, handler, &wg, errc)
 
 	// Wait for signal.
 	log.Warn().Msgf("exiting (%v)", <-errc)
@@ -118,7 +118,7 @@ func main() {
 	log.Warn().Msg("exited")
 }
 
-func Start(ctx context.Context, reader *kafka.Reader, handler *tiup.Publisher, wg *sync.WaitGroup, errc chan error) {
+func start(ctx context.Context, reader *kafka.Reader, handler *impl.Worker, wg *sync.WaitGroup, errc chan error) {
 	(*wg).Add(1)
 	go func() {
 		defer (*wg).Done()

diff --git a/publisher/design/design.go b/publisher/design/design.go
@@ -31,7 +31,7 @@ var _ = Service("tiup", func() {
 	Method("request-to-publish", func() {
 		Payload(func() {
 			Attribute("artifact_url", String, func() {
-				Description("The full url of the pushed image, contain the tag part. It will parse the repo from it.")
+				Description("The full url of the pushed OCI artifact, contain the tag part. It will parse the repo from it.")
 			})
 			Attribute("version", String, func() {
 				Description("Force set the version. Default is the artifact version read from `org.opencontainers.image.version` of the manifest config.")
@@ -65,3 +65,36 @@ var _ = Service("tiup", func() {
 		})
 	})
 })
+
+var _ = Service("fileserver", func() {
+	Description("Publisher service for static file server ")
+	HTTP(func() {
+		Path("/fs")
+	})
+	Method("request-to-publish", func() {
+		Payload(func() {
+			Attribute("artifact_url", String, func() {
+				Description("The full url of the pushed OCI artifact, contain the tag part. It will parse the repo from it.")
+			})
+			Required("artifact_url")
+		})
+		Result(ArrayOf(String), "request track ids")
+		HTTP(func() {
+			POST("/publish-request")
+			Response(StatusOK)
+		})
+	})
+	Method("query-publishing-status", func() {
+		Payload(func() {
+			Attribute("request_id", String, "request track id")
+			Required("request_id")
+		})
+		Result(String, "request state", func() {
+			Enum("queued", "processing", "success", "failed", "canceled")
+		})
+		HTTP(func() {
+			GET("/publish-request/{request_id}")
+			Response(StatusOK)
+		})
+	})
+})
diff --git a/publisher/gen/fileserver/client.go b/publisher/gen/fileserver/client.go
diff --git a/publisher/gen/fileserver/endpoints.go b/publisher/gen/fileserver/endpoints.go
diff --git a/publisher/gen/fileserver/service.go b/publisher/gen/fileserver/service.go