Implement initialization of piped service client in k8s plugin

pkg/app/pipedv1/plugin/kubernetes/server.go

@@ -16,22 +16,28 @@
 
 import (
 	"context"
+	"net"
+	"strconv"
 	"time"
 
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes/deployment"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/toolregistry"
 	"github.com/pipe-cd/pipecd/pkg/cli"
+	"github.com/pipe-cd/pipecd/pkg/plugin/logpersister"
+	"github.com/pipe-cd/pipecd/pkg/plugin/pipedapi"
 	"github.com/pipe-cd/pipecd/pkg/rpc"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
 
 type server struct {
-	apiPort     int
-	gracePeriod time.Duration
-	tls         bool
-	certFile    string
-	keyFile     string
+	apiPort                int
+	pipedPluginServicePort int
+	gracePeriod            time.Duration
+	tls                    bool
+	certFile               string
+	keyFile                string
 
 	enableGRPCReflection bool
 }
@@ -49,6 +55,8 @@
 	}
 
 	cmd.Flags().IntVar(&s.apiPort, "api-port", s.apiPort, "The port number used to run a grpc server for external apis.")
+	cmd.Flags().IntVar(&s.pipedPluginServicePort, "piped-plugin-service-port", s.pipedPluginServicePort, "The port number used to connect to the piped plugin service.") // TODO: we should discuss about the name of this flag, or we should use environment variable instead.
+	cmd.MarkFlagRequired("piped-plugin-service-port")
 	cmd.Flags().DurationVar(&s.gracePeriod, "grace-period", s.gracePeriod, "How long to wait for graceful shutdown.")
 
 	cmd.Flags().BoolVar(&s.tls, "tls", s.tls, "Whether running the gRPC server with TLS or not.")
@@ -68,15 +76,19 @@
 
 	group, ctx := errgroup.WithContext(ctx)
 
-	// TODO: create the piped plugin gRPC client here.
+	pipedapiClient, err := pipedapi.NewClient(ctx, net.JoinHostPort("localhost", strconv.Itoa(s.pipedPluginServicePort)), nil)
+	if err != nil {
+		input.Logger.Error("failed to create piped plugin service client", zap.Error(err))
+		return err
+	}
 
 	// Start a gRPC server for handling external API requests.
 	{
 		var (
 			service = deployment.NewDeploymentService(
 				input.Logger,
-				nil, // TODO: set the tool registry client here.
-				nil, // TODO: set the log persister here.
+				toolregistry.NewToolRegistry(pipedapiClient),
+				logpersister.NewPersister(pipedapiClient, input.Logger),
 			)
 			opts = []rpc.Option{
 				rpc.WithPort(s.apiPort),

pkg/app/pipedv1/plugin/toolregistry/toolregistry.go

@@ -24,6 +24,12 @@
 	client service.PluginServiceClient
 }
 
+func NewToolRegistry(client service.PluginServiceClient) *ToolRegistry {
+	return &ToolRegistry{
+		client: client,
+	}
+}
+
 func (r *ToolRegistry) InstallTool(ctx context.Context, name, version, script string) (path string, err error) {
 	res, err := r.client.InstallTool(ctx, &service.InstallToolRequest{
 		Name:          name,

pkg/plugin/pipedapi/client.go

@@ -0,0 +1,61 @@
+// 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 planner provides a piped component
+// that decides the deployment pipeline of a deployment.
+// The planner bases on the changes from git commits
+// then builds the deployment manifests to know the behavior of the deployment.
+// From that behavior the planner can decides which pipeline should be applied.
+
+package pipedapi
+
+import (
+	"context"
+	"slices"
+
+	"google.golang.org/grpc"
+
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/cmd/piped/service"
+	"github.com/pipe-cd/pipecd/pkg/rpc/rpcclient"
+)
+
+type PipedServiceClient struct {
+	service.PluginServiceClient
+	conn *grpc.ClientConn
+}
+
+func NewClient(ctx context.Context, address string, opts ...rpcclient.DialOption) (*PipedServiceClient, error) {
+	// Clone the opts to avoid modifying the original opts slice.
+	opts = slices.Clone(opts)
+
+	// Append the required options.
+	// The WithBlock option is required to make the client wait until the connection is up.
+	// The WithInsecure option is required to disable the transport security.
+	// The piped service does not require transport security because it is only used in localhost.
+	opts = append(opts, rpcclient.WithBlock(), rpcclient.WithInsecure())
+
+	conn, err := rpcclient.DialContext(ctx, address, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &PipedServiceClient{
+		PluginServiceClient: service.NewPluginServiceClient(conn),
+		conn:                conn,
+	}, nil
+}
+
+func (c *PipedServiceClient) Close() error {
+	return c.conn.Close()
+}