-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #24 from scality/feature/COSI-44-instroduce-rpc-fa…
…ctory COSI-44: Integrate Local gRPC Provisioner Server, and Refactor COSI Driver for Improved Maintainability and Observability
- Loading branch information
Showing
15 changed files
with
682 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| Copyright 2024 Scality, Inc. | ||
| 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 grpcfactory | ||
|
|
||
| import ( | ||
| "google.golang.org/grpc" | ||
| cosi "sigs.k8s.io/container-object-storage-interface-spec" | ||
| ) | ||
|
|
||
| var ( | ||
| _ cosi.IdentityClient = &COSIProvisionerClient{} | ||
| _ cosi.ProvisionerClient = &COSIProvisionerClient{} | ||
| ) | ||
|
|
||
| type COSIProvisionerClient struct { | ||
| address string | ||
| conn *grpc.ClientConn | ||
| cosi.IdentityClient | ||
| cosi.ProvisionerClient | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| package grpcfactory_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "net" | ||
| "os" | ||
| "time" | ||
|
|
||
| . "github.com/onsi/ginkgo/v2" | ||
| . "github.com/onsi/gomega" | ||
| "github.com/scality/cosi-driver/pkg/grpcfactory" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
| cosi "sigs.k8s.io/container-object-storage-interface-spec" | ||
| ) | ||
|
|
||
| type MockIdentityServer struct { | ||
| cosi.UnimplementedIdentityServer | ||
| } | ||
|
|
||
| type MockProvisionerServer struct { | ||
| cosi.UnimplementedProvisionerServer | ||
| } | ||
|
|
||
| var _ = Describe("gRPC Factory Client", func() { | ||
| var ( | ||
| client *grpcfactory.COSIProvisionerClient | ||
| grpcServer *grpc.Server | ||
| listener net.Listener | ||
| address string | ||
| ) | ||
|
|
||
| BeforeEach(func() { | ||
| address = "unix:///tmp/test.sock" | ||
| grpcServer = grpc.NewServer() | ||
|
|
||
| // Remove any existing socket file to avoid "address already in use" errors | ||
| _ = os.Remove(address[7:]) | ||
|
|
||
| // Create the listener | ||
| var err error | ||
| listener, err = net.Listen("unix", address[7:]) | ||
| Expect(err).NotTo(HaveOccurred(), "Failed to create Unix listener for gRPC server") | ||
|
|
||
| // Register mock servers | ||
| cosi.RegisterIdentityServer(grpcServer, &MockIdentityServer{}) | ||
| cosi.RegisterProvisionerServer(grpcServer, &MockProvisionerServer{}) | ||
|
|
||
| // Start the gRPC server in a separate goroutine | ||
| go func() { | ||
| err := grpcServer.Serve(listener) | ||
| if err != nil && err != grpc.ErrServerStopped { | ||
| GinkgoWriter.Println("gRPC server encountered an error:", err) | ||
| } | ||
| }() | ||
| }) | ||
|
|
||
| AfterEach(func() { | ||
| // Stop the gRPC server and close the listener | ||
| grpcServer.Stop() | ||
| if listener != nil { | ||
| listener.Close() | ||
| } | ||
| // Remove the Unix socket file to clean up after each test | ||
| _ = os.Remove(address[7:]) | ||
| }) | ||
|
|
||
| Describe("Initialization and Connection", func() { | ||
| It("should initialize and connect COSIProvisionerClient", func() { | ||
| ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||
| defer cancel() | ||
|
|
||
| // Add insecure credentials to the dial options for Unix socket | ||
| dialOpts := []grpc.DialOption{ | ||
| grpc.WithTransportCredentials(insecure.NewCredentials()), | ||
| } | ||
|
|
||
| var err error | ||
| client, err = grpcfactory.NewCOSIProvisionerClient(ctx, address, dialOpts, nil) | ||
| Expect(err).NotTo(HaveOccurred()) | ||
| Expect(client).NotTo(BeNil()) | ||
| Expect(client.IdentityClient).NotTo(BeNil()) | ||
| Expect(client.ProvisionerClient).NotTo(BeNil()) | ||
| }) | ||
| }) | ||
|
|
||
| Describe("Interface Implementation", func() { | ||
| It("should implement cosi.IdentityClient and cosi.ProvisionerClient interfaces", func() { | ||
| client = &grpcfactory.COSIProvisionerClient{ | ||
| IdentityClient: cosi.NewIdentityClient(nil), | ||
| ProvisionerClient: cosi.NewProvisionerClient(nil), | ||
| } | ||
|
|
||
| var _ cosi.IdentityClient = client | ||
| var _ cosi.ProvisionerClient = client | ||
| }) | ||
| }) | ||
|
|
||
| Describe("Interceptor Usage", func() { | ||
| It("should use ApiLogger as an interceptor if debug is true", func() { | ||
| ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||
| defer cancel() | ||
|
|
||
| client, err := grpcfactory.NewDefaultCOSIProvisionerClient(ctx, address, true) | ||
| Expect(err).NotTo(HaveOccurred()) | ||
| Expect(client).NotTo(BeNil()) | ||
| }) | ||
|
|
||
| It("should initialize without interceptors if debug is false", func() { | ||
| ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||
| defer cancel() | ||
|
|
||
| client, err := grpcfactory.NewDefaultCOSIProvisionerClient(ctx, address, false) | ||
| Expect(err).NotTo(HaveOccurred()) | ||
| Expect(client).NotTo(BeNil()) | ||
| }) | ||
| }) | ||
|
|
||
| Describe("Error Handling", func() { | ||
| It("should return an error if given an invalid address", func() { | ||
| ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||
| defer cancel() | ||
|
|
||
| // Attempt to connect using an invalid address format | ||
| _, err := grpcfactory.NewCOSIProvisionerClient(ctx, "invalid-address", nil, nil) | ||
| Expect(err).To(HaveOccurred()) | ||
| Expect(err.Error()).To(ContainSubstring("unsupported scheme")) | ||
| }) | ||
| }) | ||
| }) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| /* | ||
| Copyright 2024 Scality, Inc. | ||
| 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 grpcfactory | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "net/url" | ||
| "time" | ||
|
|
||
| "github.com/pkg/errors" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/backoff" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
| "k8s.io/klog/v2" | ||
| cosi "sigs.k8s.io/container-object-storage-interface-spec" | ||
| ) | ||
|
|
||
| const ( | ||
| maxGrpcBackoff = 5 * 30 * time.Second | ||
| grpcDialTimeout = 30 * time.Second | ||
| ) | ||
|
|
||
| func NewDefaultCOSIProvisionerClient(ctx context.Context, address string, debug bool) (*COSIProvisionerClient, error) { | ||
| backoffConfiguration := backoff.DefaultConfig | ||
| backoffConfiguration.MaxDelay = maxGrpcBackoff | ||
| dialOpts := []grpc.DialOption{ | ||
| grpc.WithTransportCredentials(insecure.NewCredentials()), // strictly restricting to local Unix domain socket | ||
| grpc.WithConnectParams(grpc.ConnectParams{ | ||
| Backoff: backoffConfiguration, | ||
| MinConnectTimeout: grpcDialTimeout, | ||
| }), | ||
| // Note: grpc.WithBlock() is deprecated; we proceed without it | ||
| } | ||
| interceptors := []grpc.UnaryClientInterceptor{} | ||
| if debug { | ||
| interceptors = append(interceptors, ApiLogger) | ||
| } | ||
| return NewCOSIProvisionerClient(ctx, address, dialOpts, interceptors) | ||
| } | ||
|
|
||
| // NewCOSIProvisionerClient creates a new GRPCClient that only supports unix domain sockets | ||
| func NewCOSIProvisionerClient(ctx context.Context, address string, dialOpts []grpc.DialOption, interceptors []grpc.UnaryClientInterceptor) (*COSIProvisionerClient, error) { | ||
| addr, err := url.Parse(address) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if addr.Scheme != "unix" { | ||
| err := fmt.Errorf("unsupported scheme: expected 'unix', found '%s'", addr.Scheme) | ||
| klog.ErrorS(err, "Invalid address scheme") | ||
| return nil, err | ||
| } | ||
| if len(interceptors) > 0 { | ||
| dialOpts = append(dialOpts, grpc.WithChainUnaryInterceptor(interceptors...)) | ||
| } | ||
| // Proceed without grpc.WithBlock(), allowing connection to establish asynchronously | ||
| conn, err := grpc.Dial(address, dialOpts...) | ||
| if err != nil { | ||
| klog.ErrorS(err, "Connection failed", "address", address) | ||
| return nil, err | ||
| } | ||
| return &COSIProvisionerClient{ | ||
| address: address, | ||
| conn: conn, | ||
| IdentityClient: cosi.NewIdentityClient(conn), | ||
| ProvisionerClient: cosi.NewProvisionerClient(conn), | ||
| }, nil | ||
| } | ||
| func NewDefaultCOSIProvisionerServer(address string, | ||
| identityServer cosi.IdentityServer, | ||
| provisionerServer cosi.ProvisionerServer) (*COSIProvisionerServer, error) { | ||
| return NewCOSIProvisionerServer(address, identityServer, provisionerServer, []grpc.ServerOption{}) | ||
| } | ||
| func NewCOSIProvisionerServer(address string, | ||
| identityServer cosi.IdentityServer, | ||
| provisionerServer cosi.ProvisionerServer, | ||
| listenOpts []grpc.ServerOption) (*COSIProvisionerServer, error) { | ||
| if identityServer == nil { | ||
| err := errors.New("Identity server cannot be nil") | ||
| klog.ErrorS(err, "Invalid argument") | ||
| return nil, err | ||
| } | ||
| if provisionerServer == nil { | ||
| err := errors.New("Provisioner server cannot be nil") | ||
| klog.ErrorS(err, "Invalid argument") | ||
| return nil, err | ||
| } | ||
| return &COSIProvisionerServer{ | ||
| address: address, | ||
| identityServer: identityServer, | ||
| provisionerServer: provisionerServer, | ||
| listenOpts: listenOpts, | ||
| }, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package grpcfactory_test | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| . "github.com/onsi/ginkgo/v2" | ||
| . "github.com/onsi/gomega" | ||
| ) | ||
|
|
||
| func TestGRPCFactorySuite(t *testing.T) { | ||
| RegisterFailHandler(Fail) | ||
| RunSpecs(t, "gRPC Factory Test Suite") | ||
| } |
Oops, something went wrong.