Implemented the backend of Metadata APIs as MetadataStoreRegistry

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

@@ -18,6 +18,7 @@ import (
 	"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 @@ type PluginAPI struct {
 	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 @@ func (a *PluginAPI) Register(server *grpc.Server) {
 	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,27 @@ func (a *PluginAPI) ReportStageLogsFromLastCheckpoint(ctx context.Context, req *
 
 	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) GetDeploymentMetadata(ctx context.Context, req *service.GetDeploymentMetadataRequest) (*service.GetDeploymentMetadataResponse, error) {
+	return a.metadataStoreRegistry.GetDeploymentMetadata(ctx, req)
+}
+
+func (a *PluginAPI) PutDeploymentMetadata(ctx context.Context, req *service.PutDeploymentMetadataRequest) (*service.PutDeploymentMetadataResponse, error) {
+	return a.metadataStoreRegistry.PutDeploymentMetadata(ctx, req)
+}
+
+func (a *PluginAPI) PutDeploymentMetadataMulti(ctx context.Context, req *service.PutDeploymentMetadataMultiRequest) (*service.PutDeploymentMetadataMultiResponse, error) {
+	return a.metadataStoreRegistry.PutDeploymentMetadataMulti(ctx, req)
+}

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

@@ -60,6 +60,7 @@ import (
 	"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/statsreporter"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/trigger"
@@ -298,10 +299,11 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) {
 		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),
@@ -388,6 +390,7 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) {
 			commandLister,
 			notifier,
 			decrypter,
+			*metadataStoreRegistry,
 			p.gracePeriod,
 			input.Logger,
 			tracerProvider,

pkg/app/pipedv1/controller/controller.go

@@ -34,6 +34,7 @@ import (
 	"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/server/service/pipedservice"
 	"github.com/pipe-cd/pipecd/pkg/git"
 	"github.com/pipe-cd/pipecd/pkg/model"
@@ -89,12 +90,13 @@ var (
 )
 
 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
 
 	// gRPC clients to communicate with plugins.
 	pluginClients []pluginapi.PluginClient
@@ -136,19 +138,21 @@ func NewController(
 	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,
-		pluginClients:    pluginClients,
-		deploymentLister: deploymentLister,
-		commandLister:    commandLister,
-		notifier:         notifier,
-		secretDecrypter:  secretDecrypter,
+		apiClient:             apiClient,
+		gitClient:             gitClient,
+		pluginClients:         pluginClients,
+		deploymentLister:      deploymentLister,
+		commandLister:         commandLister,
+		notifier:              notifier,
+		secretDecrypter:       secretDecrypter,
+		metadataStoreRegistry: metadataStoreRegistry,
 
 		planners:                              make(map[string]*planner),
 		donePlanners:                          make(map[string]time.Time),
@@ -597,6 +601,8 @@ func (c *controller) startNewScheduler(ctx context.Context, d *model.Deployment)
 		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)
@@ -614,6 +620,7 @@ func (c *controller) startNewScheduler(ctx context.Context, d *model.Deployment)
 	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 @@ import (
 
 	"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/server/service/pipedservice"
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
 	"github.com/pipe-cd/pipecd/pkg/model"
@@ -70,10 +69,8 @@ type planner struct {
 	// 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.
@@ -123,7 +120,6 @@ func newPlanner(
 		plugins:                      pluginClients,
 		apiClient:                    apiClient,
 		gitClient:                    gitClient,
-		metadataStore:                metadatastore.NewMetadataStore(apiClient, d),
 		notifier:                     notifier,
 		secretDecrypter:              secretDecrypter,
 		doneDeploymentStatus:         d.Status,
@@ -647,7 +643,7 @@ func (p *planner) reportDeploymentCancelled(ctx context.Context, commander, reas
 
 // 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 @@ import (
 
 	"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/server/service/pipedservice"
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
 	"github.com/pipe-cd/pipecd/pkg/model"
@@ -47,7 +46,6 @@ type scheduler struct {
 
 	apiClient       apiClient
 	gitClient       gitClient
-	metadataStore   metadatastore.MetadataStore
 	notifier        notifier
 	secretDecrypter secretDecrypter
 
@@ -99,7 +97,6 @@ func newScheduler(
 		stageBasedPluginsMap: stageBasedPluginsMap,
 		apiClient:            apiClient,
 		gitClient:            gitClient,
-		metadataStore:        metadatastore.NewMetadataStore(apiClient, d),
 		notifier:             notifier,
 		secretDecrypter:      secretsDecrypter,
 		doneDeploymentStatus: d.Status,
@@ -717,7 +714,7 @@ func (s *scheduler) reportDeploymentCompleted(ctx context.Context, status model.
 
 // 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
 	}

pkg/app/pipedv1/metadatastore/registry.go

@@ -0,0 +1,139 @@
+// Copyright 2024 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metadatastore
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pipe-cd/pipecd/pkg/model"
+	service "github.com/pipe-cd/pipecd/pkg/plugin/pipedservice"
+)
+
+// MetadataStoreRegistry is a registry of metadata stores for deployments.
+type MetadataStoreRegistry struct {
+	apiClient apiClient
+
+	// stores is a map of metadata store for each deployment.
+	// The key is the deployment ID.
+	stores map[string]MetadataStore
+}
+
+// NewMetadataStoreRegistry creates a new MetadataStoreRegistry.
+func NewMetadataStoreRegistry(apiClient apiClient) *MetadataStoreRegistry {
+	return &MetadataStoreRegistry{apiClient: apiClient, stores: make(map[string]MetadataStore, 0)}
+}
+
+// Register creates a new metadata store for the given deployment.
+// This must be called before other Get/Put methods are called for the deployment.
+// If the metadata store already exists, the new one will replace the existing one.
+func (r *MetadataStoreRegistry) Register(d *model.Deployment) {
+	store := NewMetadataStore(r.apiClient, d)
+	r.stores[d.Id] = store
+}
+
+// Delete deletes the metadata store for the given deployment in order to release the resources.
+// If the metadata store is not found, it is a no-op.
+func (r *MetadataStoreRegistry) Delete(deploymentID string) {
+	delete(r.stores, deploymentID)
+}
+
+// GetStageMetadata implements the backend of PluginService.GetStageMetadata().
+func (r *MetadataStoreRegistry) GetStageMetadata(ctx context.Context, req *service.GetStageMetadataRequest) (*service.GetStageMetadataResponse, error) {
+	mds, ok := r.stores[req.DeploymentId]
+	if !ok {
+		return &service.GetStageMetadataResponse{Found: false}, fmt.Errorf("metadata store not found for deployment %s", req.DeploymentId)
+	}
+
+	value, found := mds.Stage(req.StageId).Get(req.Key)
+	return &service.GetStageMetadataResponse{
+		Value: value,
+		Found: found,
+	}, nil
+}
+
+// PutStageMetadata implements the backend of PluginService.PutStageMetadata().
+func (r *MetadataStoreRegistry) PutStageMetadata(ctx context.Context, req *service.PutStageMetadataRequest) (*service.PutStageMetadataResponse, error) {
+	mds, ok := r.stores[req.DeploymentId]
+	if !ok {
+		return &service.PutStageMetadataResponse{}, fmt.Errorf("metadata store not found for deployment %s", req.DeploymentId)
+	}
+
+	err := mds.Stage(req.StageId).Put(ctx, req.Key, req.Value)
+	if err != nil {
+		return &service.PutStageMetadataResponse{}, err
+	}
+
+	return &service.PutStageMetadataResponse{}, nil
+}
+
+// PutStageMetadataMulti implements the backend of PluginService.PutStageMetadataMulti().
+func (r *MetadataStoreRegistry) PutStageMetadataMulti(ctx context.Context, req *service.PutStageMetadataMultiRequest) (*service.PutStageMetadataMultiResponse, error) {
+	mds, ok := r.stores[req.DeploymentId]
+	if !ok {
+		return &service.PutStageMetadataMultiResponse{}, fmt.Errorf("metadata store not found for deployment %s", req.DeploymentId)
+	}
+
+	err := mds.Stage(req.StageId).PutMulti(ctx, req.Metadata)
+	if err != nil {
+		return &service.PutStageMetadataMultiResponse{}, err
+	}
+
+	return &service.PutStageMetadataMultiResponse{}, nil
+}
+
+// GetDeploymentMetadata implements the backend of PluginService.GetDeploymentMetadata().
+func (r *MetadataStoreRegistry) GetDeploymentMetadata(ctx context.Context, req *service.GetDeploymentMetadataRequest) (*service.GetDeploymentMetadataResponse, error) {
+	mds, ok := r.stores[req.DeploymentId]
+	if !ok {
+		return &service.GetDeploymentMetadataResponse{Found: false}, fmt.Errorf("metadata store not found for deployment %s", req.DeploymentId)
+	}
+
+	value, found := mds.Shared().Get(req.Key)
+	return &service.GetDeploymentMetadataResponse{
+		Value: value,
+		Found: found,
+	}, nil
+}
+
+// PutDeploymentMetadata implements the backend of PluginService.PutDeploymentMetadata().
+func (r *MetadataStoreRegistry) PutDeploymentMetadata(ctx context.Context, req *service.PutDeploymentMetadataRequest) (*service.PutDeploymentMetadataResponse, error) {
+	mds, ok := r.stores[req.DeploymentId]
+	if !ok {
+		return &service.PutDeploymentMetadataResponse{}, fmt.Errorf("metadata store not found for deployment %s", req.DeploymentId)
+	}
+
+	err := mds.Shared().Put(ctx, req.Key, req.Value)
+	if err != nil {
+		return &service.PutDeploymentMetadataResponse{}, err
+	}
+
+	return &service.PutDeploymentMetadataResponse{}, nil
+}
+
+// PutDeploymentMetadataMulti implements the backend of PluginService.PutDeploymentMetadataMulti().
+func (r *MetadataStoreRegistry) PutDeploymentMetadataMulti(ctx context.Context, req *service.PutDeploymentMetadataMultiRequest) (*service.PutDeploymentMetadataMultiResponse, error) {
+	mds, ok := r.stores[req.DeploymentId]
+	if !ok {
+		return &service.PutDeploymentMetadataMultiResponse{}, fmt.Errorf("metadata store not found for deployment %s", req.DeploymentId)
+	}
+
+	err := mds.Shared().PutMulti(ctx, req.Metadata)
+	if err != nil {
+		return &service.PutDeploymentMetadataMultiResponse{}, err
+	}
+
+	return &service.PutDeploymentMetadataMultiResponse{}, nil
+}