Implemented the backend of Metadata APIs as MetadataStoreRegistry

pkg/app/pipedv1/cmd/piped/grpcapi/plugin_api.go

@@ -18,6 +18,7 @@
 	"context"
 	"fmt"
 
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/metadatastore"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
 	service "github.com/pipe-cd/pipecd/pkg/plugin/pipedservice"
@@ -32,8 +33,9 @@
 	cfg       *config.PipedSpec
 	apiClient apiClient
 
-	toolRegistry *toolRegistry
-	Logger       *zap.Logger
+	toolRegistry          *toolRegistry
+	Logger                *zap.Logger
+	metadataStoreRegistry *metadatastore.MetadataStoreRegistry
 }
 
 type apiClient interface {
@@ -46,17 +48,18 @@
 	service.RegisterPluginServiceServer(server, a)
 }
 
-func NewPluginAPI(cfg *config.PipedSpec, apiClient apiClient, toolsDir string, logger *zap.Logger) (*PluginAPI, error) {
+func NewPluginAPI(cfg *config.PipedSpec, apiClient apiClient, toolsDir string, logger *zap.Logger, metadataStoreRegistry *metadatastore.MetadataStoreRegistry) (*PluginAPI, error) {
 	toolRegistry, err := newToolRegistry(toolsDir)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create tool registry: %w", err)
 	}
 
 	return &PluginAPI{
-		cfg:          cfg,
-		apiClient:    apiClient,
-		toolRegistry: toolRegistry,
-		Logger:       logger.Named("plugin-api"),
+		cfg:                   cfg,
+		apiClient:             apiClient,
+		toolRegistry:          toolRegistry,
+		Logger:                logger.Named("plugin-api"),
+		metadataStoreRegistry: metadataStoreRegistry,
 	}, nil
 }
 
@@ -112,3 +115,31 @@
 
 	return &service.ReportStageLogsFromLastCheckpointResponse{}, nil
 }
+
+func (a *PluginAPI) GetStageMetadata(ctx context.Context, req *service.GetStageMetadataRequest) (*service.GetStageMetadataResponse, error) {
+	return a.metadataStoreRegistry.GetStageMetadata(ctx, req)
+}
+
+func (a *PluginAPI) PutStageMetadata(ctx context.Context, req *service.PutStageMetadataRequest) (*service.PutStageMetadataResponse, error) {
+	return a.metadataStoreRegistry.PutStageMetadata(ctx, req)
+}
+
+func (a *PluginAPI) PutStageMetadataMulti(ctx context.Context, req *service.PutStageMetadataMultiRequest) (*service.PutStageMetadataMultiResponse, error) {
+	return a.metadataStoreRegistry.PutStageMetadataMulti(ctx, req)
+}
+
+func (a *PluginAPI) GetDeploymentPluginMetadata(ctx context.Context, req *service.GetDeploymentPluginMetadataRequest) (*service.GetDeploymentPluginMetadataResponse, error) {
+	return a.metadataStoreRegistry.GetDeploymentPluginMetadata(ctx, req)
+}
+
+func (a *PluginAPI) PutDeploymentPluginMetadata(ctx context.Context, req *service.PutDeploymentPluginMetadataRequest) (*service.PutDeploymentPluginMetadataResponse, error) {
+	return a.metadataStoreRegistry.PutDeploymentPluginMetadata(ctx, req)
+}
+
+func (a *PluginAPI) PutDeploymentPluginMetadataMulti(ctx context.Context, req *service.PutDeploymentPluginMetadataMultiRequest) (*service.PutDeploymentPluginMetadataMultiResponse, error) {
+	return a.metadataStoreRegistry.PutDeploymentPluginMetadataMulti(ctx, req)
+}
+
+func (a *PluginAPI) GetDeploymentSharedMetadata(ctx context.Context, req *service.GetDeploymentSharedMetadataRequest) (*service.GetDeploymentSharedMetadataResponse, error) {
+	return a.metadataStoreRegistry.GetDeploymentSharedMetadata(ctx, req)
+}

pkg/app/pipedv1/cmd/piped/piped.go

@@ -60,6 +60,7 @@
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/eventwatcher"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/metadatastore"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/notifier"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/statsreporter"
@@ -299,10 +300,11 @@
 		eventLister = store.Lister()
 	}
 
+	metadataStoreRegistry := metadatastore.NewMetadataStoreRegistry(apiClient)
 	// Start running plugin service server.
 	{
 		var (
-			service, err = grpcapi.NewPluginAPI(cfg, apiClient, p.toolsDir, input.Logger)
+			service, err = grpcapi.NewPluginAPI(cfg, apiClient, p.toolsDir, input.Logger, metadataStoreRegistry)
 			opts         = []rpc.Option{
 				rpc.WithPort(p.pluginServicePort),
 				rpc.WithGracePeriod(p.gracePeriod),
@@ -400,6 +402,7 @@
 			commandLister,
 			notifier,
 			decrypter,
+			*metadataStoreRegistry,
 			p.gracePeriod,
 			input.Logger,
 			tracerProvider,

pkg/app/pipedv1/controller/controller.go

@@ -34,6 +34,7 @@
 	"google.golang.org/grpc/status"
 
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/metadatastore"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
 	"github.com/pipe-cd/pipecd/pkg/git"
@@ -88,12 +89,13 @@
 )
 
 type controller struct {
-	apiClient        apiClient
-	gitClient        gitClient
-	deploymentLister deploymentLister
-	commandLister    commandLister
-	notifier         notifier
-	secretDecrypter  secretDecrypter
+	apiClient             apiClient
+	gitClient             gitClient
+	deploymentLister      deploymentLister
+	commandLister         commandLister
+	notifier              notifier
+	secretDecrypter       secretDecrypter
+	metadataStoreRegistry metadatastore.MetadataStoreRegistry
 
 	// The registry of all plugins.
 	pluginRegistry plugin.PluginRegistry
@@ -134,19 +136,21 @@
 	commandLister commandLister,
 	notifier notifier,
 	secretDecrypter secretDecrypter,
+	metadataStoreRegistry metadatastore.MetadataStoreRegistry,
 	gracePeriod time.Duration,
 	logger *zap.Logger,
 	tracerProvider trace.TracerProvider,
 ) DeploymentController {
 
 	return &controller{
-		apiClient:        apiClient,
-		gitClient:        gitClient,
-		pluginRegistry:   pluginRegistry,
-		deploymentLister: deploymentLister,
-		commandLister:    commandLister,
-		notifier:         notifier,
-		secretDecrypter:  secretDecrypter,
+		apiClient:             apiClient,
+		gitClient:             gitClient,
+		pluginRegistry:        pluginRegistry,
+		deploymentLister:      deploymentLister,
+		commandLister:         commandLister,
+		notifier:              notifier,
+		secretDecrypter:       secretDecrypter,
+		metadataStoreRegistry: metadataStoreRegistry,
 
 		planners:                              make(map[string]*planner),
 		donePlanners:                          make(map[string]time.Time),
@@ -577,6 +581,8 @@
 		c.tracerProvider,
 	)
 
+	c.metadataStoreRegistry.Register(d)
+
 	cleanup := func() {
 		logger.Info("cleaning up working directory for scheduler", zap.String("working-dir", workingDir))
 		err := os.RemoveAll(workingDir)
@@ -594,6 +600,7 @@
 	go func() {
 		defer c.wg.Done()
 		defer cleanup()
+		defer c.metadataStoreRegistry.Delete(d.Id)
 		if err := scheduler.Run(ctx); err != nil {
 			logger.Error("failed to run scheduler", zap.Error(err))
 		}

pkg/app/pipedv1/controller/planner.go

@@ -30,7 +30,6 @@
 
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/deploysource"
-	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/metadatastore"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
@@ -64,10 +63,8 @@
 	// The gitClient is used to perform git commands.
 	gitClient gitClient
 
-	// The notifier and metadataStore are used for
-	// notification features.
-	notifier      notifier
-	metadataStore metadatastore.MetadataStore
+	// The notifier is used for notification features.
+	notifier notifier
 
 	// The secretDecrypter is used to decrypt secrets
 	// which encrypted using PipeCD built-in secret management.
@@ -118,7 +115,6 @@
 		pluginRegistry:               pluginRegistry,
 		apiClient:                    apiClient,
 		gitClient:                    gitClient,
-		metadataStore:                metadatastore.NewMetadataStore(apiClient, d),
 		notifier:                     notifier,
 		secretDecrypter:              secretDecrypter,
 		doneDeploymentStatus:         d.Status,
@@ -657,7 +653,7 @@
 
 // getApplicationNotificationMentions returns the list of users groups who should be mentioned in the notification.
 func (p *planner) getApplicationNotificationMentions(event model.NotificationEventType) ([]string, []string, error) {
-	n, ok := p.metadataStore.Shared().Get(model.MetadataKeyDeploymentNotification)
+	n, ok := p.deployment.Metadata[model.MetadataKeyDeploymentNotification]
 	if !ok {
 		return []string{}, []string{}, nil
 	}

pkg/app/pipedv1/controller/scheduler.go

@@ -30,7 +30,6 @@
 
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/deploysource"
-	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/metadatastore"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
@@ -47,7 +46,6 @@
 
 	apiClient       apiClient
 	gitClient       gitClient
-	metadataStore   metadatastore.MetadataStore
 	notifier        notifier
 	secretDecrypter secretDecrypter
 
@@ -99,7 +97,6 @@
 		apiClient:            apiClient,
 		gitClient:            gitClient,
 		pluginRegistry:       pluginRegistry,
-		metadataStore:        metadatastore.NewMetadataStore(apiClient, d),
 		notifier:             notifier,
 		secretDecrypter:      secretsDecrypter,
 		doneDeploymentStatus: d.Status,
@@ -717,7 +714,7 @@
 
 // getApplicationNotificationMentions returns the list of users groups who should be mentioned in the notification.
 func (s *scheduler) getApplicationNotificationMentions(event model.NotificationEventType) ([]string, []string, error) {
-	n, ok := s.metadataStore.Shared().Get(model.MetadataKeyDeploymentNotification)
+	n, ok := s.deployment.Metadata[model.MetadataKeyDeploymentNotification]
 	if !ok {
 		return []string{}, []string{}, nil
 	}