Implement the tool registry in the piped side

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

@@ -17,6 +17,7 @@
 import (
 	"context"
 	"fmt"
+	"os"
 
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/cmd/piped/service"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
@@ -34,7 +35,8 @@
 	cfg       *config.PipedSpec
 	apiClient apiClient
 
-	Logger *zap.Logger
+	toolRegistry *toolRegistry
+	Logger       *zap.Logger
 }
 
 type apiClient interface {
@@ -47,12 +49,18 @@
 	service.RegisterPluginServiceServer(server, a)
 }
 
-func NewPluginAPI(cfg *config.PipedSpec, apiClient apiClient, logger *zap.Logger) *PluginAPI {
-	return &PluginAPI{
-		cfg:       cfg,
-		apiClient: apiClient,
-		Logger:    logger.Named("plugin-api"),
+func NewPluginAPI(cfg *config.PipedSpec, apiClient apiClient, toolsDir string, logger *zap.Logger) (*PluginAPI, error) {
+	toolRegistry, err := newToolRegistry(toolsDir, os.TempDir())
+	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"),
+	}, nil
 }
 
 func (a *PluginAPI) DecryptSecret(ctx context.Context, req *service.DecryptSecretRequest) (*service.DecryptSecretResponse, error) {
@@ -80,6 +88,22 @@
 	}, nil
 }
 
+// InstallTool installs the given tool.
+// installed binary's filename becomes `name-version`.
+func (a *PluginAPI) InstallTool(ctx context.Context, req *service.InstallToolRequest) (*service.InstallToolResponse, error) {
+	p, err := a.toolRegistry.InstallTool(ctx, req.GetName(), req.GetVersion(), req.GetInstallScript())
+	if err != nil {
+		a.Logger.Error("failed to install tool",
+			zap.String("name", req.GetName()),
+			zap.String("version", req.GetVersion()),
+			zap.Error(err))
+		return nil, err
+	}
+	return &service.InstallToolResponse{
+		InstalledPath: p,
+	}, nil
+}
+
 func (a *PluginAPI) ReportStageLogs(ctx context.Context, req *service.ReportStageLogsRequest) (*service.ReportStageLogsResponse, error) {
 	_, err := a.apiClient.ReportStageLogs(ctx, &pipedservice.ReportStageLogsRequest{
 		DeploymentId: req.DeploymentId,

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

@@ -0,0 +1,133 @@
+// 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 grpcapi
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"text/template"
+
+	"golang.org/x/sync/singleflight"
+)
+
+type toolRegistry struct {
+	toolsDir string
+	tmpDir   string
+	group    singleflight.Group
+}
+
+type templateValues struct {
+	Name    string
+	Version string
+	OutPath string
+	TmpDir  string
+	Arch    string
+	Os      string
+}
+
+func newToolRegistry(toolsDir, tmpDir string) (*toolRegistry, error) {
+	tmpDir, err := os.MkdirTemp(tmpDir, "tool-registry")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create a temporary directory: %w", err)
+	}
+	return &toolRegistry{
+		toolsDir: toolsDir,
+		tmpDir:   tmpDir,
+	}, nil
+}
+
+func (r *toolRegistry) newTmpDir() (string, error) {
+	return os.MkdirTemp(r.tmpDir, "")
+}
+
+func (r *toolRegistry) outPath() (string, error) {
+	target, err := r.newTmpDir()
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(target, "out"), nil
+}
+
+func (r *toolRegistry) InstallTool(ctx context.Context, name, version, script string) (string, error) {
+	out, err, _ := r.group.Do(fmt.Sprintf("%s-%s", name, version), func() (interface{}, error) {
+		return r.installTool(ctx, name, version, script)
+	})
+	if err != nil {
+		return "", fmt.Errorf("failed to install the tool %s-%s: %w", name, version, err)
+	}
+	return out.(string), nil // the result is always string
+}
+
+func (r *toolRegistry) installTool(ctx context.Context, name, version, script string) (path string, err error) {
+	outPath, err := r.outPath()
+	if err != nil {
+		return "", err
+	}
+
+	tmpDir, err := r.newTmpDir()
+	if err != nil {
+		return "", err
+	}
+
+	t, err := template.New("install script").Parse(script)
+	if err != nil {
+		return "", err
+	}
+
+	vars := templateValues{
+		Name:    name,
+		Version: version,
+		OutPath: outPath,
+		TmpDir:  tmpDir,
+		Arch:    runtime.GOARCH,
+		Os:      runtime.GOOS,
+	}
+	var buf bytes.Buffer
+	if err := t.Execute(&buf, vars); err != nil {
+		return "", err
+	}
+
+	cmd := exec.CommandContext(ctx, "/bin/sh", "-c", buf.String())
+	if out, err := cmd.CombinedOutput(); err != nil {
+		return "", fmt.Errorf("failed to execute the install script: %w, output: %s", err, string(out))
+	}
+
+	if err := os.Chmod(outPath, 0o755); err != nil {
+		return "", err
+	}
+
+	target := filepath.Join(r.toolsDir, fmt.Sprintf("%s-%s", name, version))
+	if out, err := exec.CommandContext(ctx, "/bin/sh", "-c", "mv "+outPath+" "+target).CombinedOutput(); err != nil {
+		return "", fmt.Errorf("failed to move the installed binary: %w, output: %s", err, string(out))
+	}
+
+	if err := os.RemoveAll(tmpDir); err != nil {
+		return "", err
+	}
+
+	return target, nil
+}
+
+func (r *toolRegistry) Close() error {
+	if err := os.RemoveAll(r.tmpDir); err != nil {
+		return err
+	}
+	return nil
+}

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

@@ -0,0 +1,86 @@
+// 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 grpcapi
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestToolRegistry_InstallTool(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	toolsDir, err := os.MkdirTemp(t.TempDir(), "tools")
+	require.NoError(t, err)
+	tmpDir, err := os.MkdirTemp(t.TempDir(), "tmp")
+	require.NoError(t, err)
+
+	registry := &toolRegistry{
+		toolsDir: toolsDir,
+		tmpDir:   tmpDir,
+	}
+
+	tests := []struct {
+		name        string
+		toolName    string
+		toolVersion string
+		script      string
+		wantErr     bool
+	}{
+		{
+			name:        "valid script",
+			toolName:    "tool-a",
+			toolVersion: "1.0.0",
+			script:      `touch {{ .OutPath }}`,
+			wantErr:     false,
+		},
+		{
+			name:        "output is not found",
+			toolName:    "tool-b",
+			toolVersion: "1.0.0",
+			script:      "exit 0",
+			wantErr:     true,
+		},
+		{
+			name:        "script failed",
+			toolName:    "tool-c",
+			toolVersion: "1.0.0",
+			script:      "exit 1",
+			wantErr:     true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			out, err := registry.InstallTool(ctx, tt.toolName, tt.toolVersion, tt.script)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.FileExists(t, out)
+			assert.True(t, strings.HasSuffix(out, tt.toolName+"-"+tt.toolVersion), "output path should have the tool {name}-{version}, got %s", out)
+		})
+	}
+}

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

@@ -304,15 +304,19 @@
 	// Start running plugin service server.
 	{
 		var (
-			service = grpcapi.NewPluginAPI(cfg, apiClient, input.Logger)
-			opts    = []rpc.Option{
+			service, err = grpcapi.NewPluginAPI(cfg, apiClient, p.toolsDir, input.Logger)
+			opts         = []rpc.Option{
 				rpc.WithPort(p.pluginServicePort),
 				rpc.WithGracePeriod(p.gracePeriod),
 				rpc.WithLogger(input.Logger),
 				rpc.WithLogUnaryInterceptor(input.Logger),
 				rpc.WithRequestValidationUnaryInterceptor(),
 			}
 		)
+		if err != nil {
+			input.Logger.Error("failed to create plugin service", zap.Error(err))
+			return err
+		}
 		// TODO: Ensure piped <-> plugin communication is secure.
 		server := rpc.NewServer(service, opts...)
 		group.Go(func() error {