Implement RPC to retrieve planpreview results

cmd/pipecd/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/app/api/apikeyverifier:go_default_library",
         "//pkg/app/api/applicationlivestatestore:go_default_library",
         "//pkg/app/api/authhandler:go_default_library",
+        "//pkg/app/api/commandoutputstore:go_default_library",
         "//pkg/app/api/commandstore:go_default_library",
         "//pkg/app/api/grpcapi:go_default_library",
         "//pkg/app/api/pipedverifier:go_default_library",

cmd/pipecd/server.go

@@ -32,6 +32,7 @@ import (
 	"github.com/pipe-cd/pipe/pkg/app/api/apikeyverifier"
 	"github.com/pipe-cd/pipe/pkg/app/api/applicationlivestatestore"
 	"github.com/pipe-cd/pipe/pkg/app/api/authhandler"
+	"github.com/pipe-cd/pipe/pkg/app/api/commandoutputstore"
 	"github.com/pipe-cd/pipe/pkg/app/api/commandstore"
 	"github.com/pipe-cd/pipe/pkg/app/api/grpcapi"
 	"github.com/pipe-cd/pipe/pkg/app/api/pipedverifier"
@@ -178,6 +179,7 @@ func (s *server) run(ctx context.Context, t cli.Telemetry) error {
 	alss := applicationlivestatestore.NewStore(fs, cache, t.Logger)
 	cmds := commandstore.NewStore(ds, cache, t.Logger)
 	is := insightstore.NewStore(fs)
+	cmdOutputStore := commandoutputstore.NewStore(fs, t.Logger)
 
 	// Start a gRPC server for handling PipedAPI requests.
 	{
@@ -189,7 +191,7 @@ func (s *server) run(ctx context.Context, t cli.Telemetry) error {
 				datastore.NewPipedStore(ds),
 				t.Logger,
 			)
-			service = grpcapi.NewPipedAPI(ctx, ds, sls, alss, cmds, t.Logger)
+			service = grpcapi.NewPipedAPI(ctx, ds, sls, alss, cmds, cmdOutputStore, t.Logger)
 			opts    = []rpc.Option{
 				rpc.WithPort(s.pipedAPIPort),
 				rpc.WithGracePeriod(s.gracePeriod),
@@ -220,7 +222,7 @@ func (s *server) run(ctx context.Context, t cli.Telemetry) error {
 				datastore.NewAPIKeyStore(ds),
 				t.Logger,
 			)
-			service = grpcapi.NewAPI(ds, cmds, t.Logger)
+			service = grpcapi.NewAPI(ds, cmds, cmdOutputStore, t.Logger)
 			opts    = []rpc.Option{
 				rpc.WithPort(s.apiPort),
 				rpc.WithGracePeriod(s.gracePeriod),

pkg/app/api/commandoutputstore/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["store.go"],
+    importpath = "github.com/pipe-cd/pipe/pkg/app/api/commandoutputstore",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/filestore:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    size = "small",
+    srcs = ["store_test.go"],
+    embed = [":go_default_library"],
+)

pkg/app/api/commandoutputstore/store.go

@@ -0,0 +1,71 @@
+// Copyright 2021 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 commandoutputstore
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.uber.org/zap"
+
+	"github.com/pipe-cd/pipe/pkg/filestore"
+)
+
+var (
+	ErrNotFound = errors.New("not found")
+)
+
+type Store interface {
+	Get(ctx context.Context, commandID string) ([]byte, error)
+	Put(ctx context.Context, commandID string, data []byte) error
+}
+
+type store struct {
+	backend filestore.Store
+	logger  *zap.Logger
+}
+
+func NewStore(fs filestore.Store, logger *zap.Logger) Store {
+	return &store{
+		backend: fs,
+		logger:  logger.Named("command-output-store"),
+	}
+}
+
+func (s *store) Get(ctx context.Context, commandID string) ([]byte, error) {
+	path := dataPath(commandID)
+	obj, err := s.backend.GetObject(ctx, path)
+	if err != nil {
+		if err == filestore.ErrNotFound {
+			return nil, ErrNotFound
+		}
+		s.logger.Error("failed to get command ouput from filestore",
+			zap.String("command", commandID),
+			zap.Error(err),
+		)
+		return nil, err
+	}
+	return obj.Content, nil
+}
+
+func (s *store) Put(ctx context.Context, commandID string, data []byte) error {
+	path := dataPath(commandID)
+	return s.backend.PutObject(ctx, path, data)
+}
+
+func dataPath(commandID string) string {
+	return fmt.Sprintf("command-output/%s.json", commandID)
+}

pkg/app/api/commandoutputstore/store_test.go

@@ -0,0 +1,15 @@
+// Copyright 2021 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 commandoutputstore

pkg/app/api/grpcapi/BUILD.bazel

@@ -5,8 +5,8 @@ go_library(
     srcs = [
         "api.go",
         "deployment_config_templates.go",
+        "grpcapi.go",
         "piped_api.go",
-        "utils.go",
         "web_api.go",
         ":deployment_config_templates.embed",  #keep
     ],

pkg/app/api/grpcapi/api.go

@@ -16,8 +16,10 @@ package grpcapi
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/google/uuid"
 	"go.uber.org/zap"
@@ -34,12 +36,13 @@ import (
 
 // API implements the behaviors for the gRPC definitions of API.
 type API struct {
-	applicationStore datastore.ApplicationStore
-	environmentStore datastore.EnvironmentStore
-	deploymentStore  datastore.DeploymentStore
-	pipedStore       datastore.PipedStore
-	eventStore       datastore.EventStore
-	commandStore     commandstore.Store
+	applicationStore    datastore.ApplicationStore
+	environmentStore    datastore.EnvironmentStore
+	deploymentStore     datastore.DeploymentStore
+	pipedStore          datastore.PipedStore
+	eventStore          datastore.EventStore
+	commandStore        commandstore.Store
+	commandOutputGetter commandOutputGetter
 
 	logger *zap.Logger
 }
@@ -48,16 +51,18 @@ type API struct {
 func NewAPI(
 	ds datastore.DataStore,
 	cmds commandstore.Store,
+	cog commandOutputGetter,
 	logger *zap.Logger,
 ) *API {
 	a := &API{
-		applicationStore: datastore.NewApplicationStore(ds),
-		environmentStore: datastore.NewEnvironmentStore(ds),
-		deploymentStore:  datastore.NewDeploymentStore(ds),
-		pipedStore:       datastore.NewPipedStore(ds),
-		eventStore:       datastore.NewEventStore(ds),
-		commandStore:     cmds,
-		logger:           logger.Named("api"),
+		applicationStore:    datastore.NewApplicationStore(ds),
+		environmentStore:    datastore.NewEnvironmentStore(ds),
+		deploymentStore:     datastore.NewDeploymentStore(ds),
+		pipedStore:          datastore.NewPipedStore(ds),
+		eventStore:          datastore.NewEventStore(ds),
+		commandStore:        cmds,
+		commandOutputGetter: cog,
+		logger:              logger.Named("api"),
 	}
 	return a
 }
@@ -376,17 +381,62 @@ func (a *API) RequestPlanPreview(ctx context.Context, req *apiservice.RequestPla
 }
 
 func (a *API) GetPlanPreviewResults(ctx context.Context, req *apiservice.GetPlanPreviewResultsRequest) (*apiservice.GetPlanPreviewResultsResponse, error) {
-	_, err := requireAPIKey(ctx, model.APIKey_READ_WRITE, a.logger)
+	key, err := requireAPIKey(ctx, model.APIKey_READ_WRITE, a.logger)
 	if err != nil {
 		return nil, err
 	}
 
-	// TODO: Implement GetPlanPreviewResults RPC.
+	const freshDuration = 24 * time.Hour
 
-	// 1. Check whether the command is handled or not.
-	// 2. Retrieve command data from filestore.
+	// Validate based on command model stored in datastore.
+	for _, commandID := range req.Commands {
+		cmd, err := getCommand(ctx, a.commandStore, commandID, a.logger)
+		if err != nil {
+			return nil, err
+		}
+		if cmd.ProjectId != key.ProjectId {
+			a.logger.Warn("detected a request to get planpreview result of an unowned command",
+				zap.String("command", commandID),
+				zap.String("command-project-id", cmd.ProjectId),
+				zap.String("request-project-id", key.ProjectId),
+			)
+			return nil, status.Error(codes.PermissionDenied, fmt.Sprintf("The requested command %s does not belong to your project", commandID))
+		}
+		if cmd.Type != model.Command_BUILD_PLAN_PREVIEW {
+			return nil, status.Error(codes.FailedPrecondition, fmt.Sprint("Command %s is not a plan preview command", commandID))
+		}
+		if !cmd.IsHandled() {
+			return nil, status.Error(codes.FailedPrecondition, fmt.Sprintf("Command %s is not completed yet", commandID))
+		}
+		// There is no reason to fetch output data of command that has been completed a long time ago.
+		// So in order to prevent unintended actions, we disallow that ability.
+		if time.Since(time.Unix(cmd.HandledAt, 0)) > freshDuration {
+			return nil, status.Error(codes.FailedPrecondition, fmt.Sprintf("The output data for command %s is too old for access", commandID))
+		}
+	}
+
+	results := make([]*model.ApplicationPlanPreviewResult, 0, len(req.Commands))
+
+	// Fetch ouput data to build results.
+	for _, commandID := range req.Commands {
+		data, err := a.commandOutputGetter.Get(ctx, commandID)
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to retrieve output data of command %s", commandID))
+		}
+		var result model.ApplicationPlanPreviewResult
+		if err := json.Unmarshal(data, &result); err != nil {
+			a.logger.Error("failed to unmarshal applicaiton plan preview result",
+				zap.String("command", commandID),
+				zap.Error(err),
+			)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to decode output data of command %s", commandID))
+		}
+		results = append(results, &result)
+	}
 
-	return &apiservice.GetPlanPreviewResultsResponse{}, nil
+	return &apiservice.GetPlanPreviewResultsResponse{
+		Results: results,
+	}, nil
 }
 
 // requireAPIKey checks the existence of an API key inside the given context

pkg/app/api/grpcapi/utils.go → pkg/app/api/grpcapi/grpcapi.go

@@ -1,4 +1,4 @@
-// Copyright 2020 The PipeCD Authors.
+// Copyright 2021 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.
@@ -28,6 +28,14 @@ import (
 	"github.com/pipe-cd/pipe/pkg/model"
 )
 
+type commandOutputGetter interface {
+	Get(ctx context.Context, commandID string) ([]byte, error)
+}
+
+type commandOutputPutter interface {
+	Put(ctx context.Context, commandID string, data []byte) error
+}
+
 func getPiped(ctx context.Context, store datastore.PipedStore, id string, logger *zap.Logger) (*model.Piped, error) {
 	piped, err := store.GetPiped(ctx, id)
 	if errors.Is(err, datastore.ErrNotFound) {

pkg/app/api/grpcapi/piped_api.go

@@ -47,6 +47,7 @@ type PipedAPI struct {
 	stageLogStore             stagelogstore.Store
 	applicationLiveStateStore applicationlivestatestore.Store
 	commandStore              commandstore.Store
+	commandOutputPutter       commandOutputPutter
 
 	appPipedCache        cache.Cache
 	deploymentPipedCache cache.Cache
@@ -56,7 +57,7 @@ type PipedAPI struct {
 }
 
 // NewPipedAPI creates a new PipedAPI instance.
-func NewPipedAPI(ctx context.Context, ds datastore.DataStore, sls stagelogstore.Store, alss applicationlivestatestore.Store, cs commandstore.Store, logger *zap.Logger) *PipedAPI {
+func NewPipedAPI(ctx context.Context, ds datastore.DataStore, sls stagelogstore.Store, alss applicationlivestatestore.Store, cs commandstore.Store, cop commandOutputPutter, logger *zap.Logger) *PipedAPI {
 	a := &PipedAPI{
 		applicationStore:          datastore.NewApplicationStore(ds),
 		deploymentStore:           datastore.NewDeploymentStore(ds),
@@ -68,6 +69,7 @@ func NewPipedAPI(ctx context.Context, ds datastore.DataStore, sls stagelogstore.
 		stageLogStore:             sls,
 		applicationLiveStateStore: alss,
 		commandStore:              cs,
+		commandOutputPutter:       cop,
 		appPipedCache:             memorycache.NewTTLCache(ctx, 24*time.Hour, 3*time.Hour),
 		deploymentPipedCache:      memorycache.NewTTLCache(ctx, 24*time.Hour, 3*time.Hour),
 		envProjectCache:           memorycache.NewTTLCache(ctx, 24*time.Hour, 3*time.Hour),
@@ -616,7 +618,15 @@ func (a *PipedAPI) ReportCommandHandled(ctx context.Context, req *pipedservice.R
 		return nil, status.Error(codes.PermissionDenied, "The current piped does not have requested command")
 	}
 
-	// TODO: Store the additional data of command into the filestore before marking the command as handled.
+	if len(req.Output) > 0 {
+		if err := a.commandOutputPutter.Put(ctx, req.CommandId, req.Output); err != nil {
+			a.logger.Error("failed to store output of command",
+				zap.String("command_id", req.CommandId),
+				zap.Error(err),
+			)
+			return nil, status.Error(codes.Internal, "Failed to store output of command")
+		}
+	}
 
 	err = a.commandStore.UpdateCommandHandled(ctx, req.CommandId, req.Status, req.Metadata, req.HandledAt)
 	if err != nil {

pkg/app/api/service/pipedservice/service.proto

@@ -340,8 +340,8 @@ message ReportCommandHandledRequest {
     pipe.model.CommandStatus status = 2 [(validate.rules).enum.defined_only = true];
     map<string,string> metadata = 3;
     int64 handled_at = 4 [(validate.rules).int64.gt = 0];
-    // Additional data to be stored in filestore.
-    bytes data = 5;
+    // Additional output data to be stored in filestore.
+    bytes output = 5;
 }
 
 message ReportCommandHandledResponse {

pkg/model/command.go

@@ -20,3 +20,7 @@ type ReportableCommand struct {
 	*Command
 	Report func(ctx context.Context, status CommandStatus, metadata map[string]string) error
 }
+
+func (c *Command) IsHandled() bool {
+	return c.Status != CommandStatus_COMMAND_NOT_HANDLED_YET
+}