From 84a1cded40d0ac8881f5ab0fe26cf7d44ee706ef Mon Sep 17 00:00:00 2001 From: Damian Czaja Date: Wed, 25 Sep 2024 16:00:35 +0200 Subject: [PATCH] KUBE-556: get GCP token source at startup (#14) --- cmd/proxy/main.go | 16 +- go.mod | 2 - go.sum | 4 - internal/cloud/gcp/executor.go | 21 +-- internal/cloud/gcp/gcpauth/client.go | 38 +---- internal/proxy/client.go | 52 ++++--- internal/proxy/client_test.go | 211 ++++++++++++++------------- 7 files changed, 165 insertions(+), 179 deletions(-) diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index 720a548..a997b58 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -3,7 +3,6 @@ package main import ( "context" "fmt" - "net/http" "path" "runtime" "time" @@ -13,7 +12,6 @@ import ( "google.golang.org/grpc/backoff" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/metadata" "cloud-proxy/internal/cloud/gcp" "cloud-proxy/internal/cloud/gcp/gcpauth" @@ -33,6 +31,13 @@ func main() { cfg := config.Get() logger := setupLogger(cfg) + ctx := context.Background() + + tokenSource, err := gcpauth.NewTokenSource(ctx) + if err != nil { + logger.WithError(err).Panicf("Failed to create GCP credentials source") + } + dialOpts := make([]grpc.DialOption, 0) if cfg.CastAI.DisableGRPCTLS { // ONLY For testing purposes. @@ -72,14 +77,11 @@ func main() { } }(conn) - ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs( - "authorization", fmt.Sprintf("Token %s", cfg.CastAI.APIKey), - )) + client := proxy.New(conn, gcp.New(tokenSource), logger, + cfg.GetPodName(), cfg.ClusterID, GetVersion(), cfg.CastAI.APIKey, cfg.KeepAlive, cfg.KeepAliveTimeout) go startHealthServer(logger, cfg.HealthAddress) - client := proxy.New(conn, gcp.New(gcpauth.NewCredentialsSource(), http.DefaultClient), logger, - cfg.GetPodName(), cfg.ClusterID, GetVersion(), cfg.KeepAlive, cfg.KeepAliveTimeout) err = client.Run(ctx) if err != nil { logger.Panicf("Failed to run client: %v", err) diff --git a/go.mod b/go.mod index be4da2e..cbd3857 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,6 @@ module cloud-proxy go 1.23.1 require ( - cloud.google.com/go/compute v1.28.0 cloud.google.com/go/container v1.39.0 github.com/golang/mock v1.6.0 github.com/google/uuid v1.6.0 @@ -56,7 +55,6 @@ require ( golang.org/x/sys v0.25.0 // indirect golang.org/x/text v0.18.0 // indirect golang.org/x/time v0.6.0 // indirect - google.golang.org/genproto v0.0.0-20240903143218-8af14fe29dc1 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20240827150818-7e3bb234dfed // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect gopkg.in/ini.v1 v1.67.0 // indirect diff --git a/go.sum b/go.sum index d5f294e..7156d28 100644 --- a/go.sum +++ b/go.sum @@ -5,8 +5,6 @@ cloud.google.com/go/auth v0.9.3 h1:VOEUIAADkkLtyfr3BLa3R8Ed/j6w1jTBmARx+wb5w5U= cloud.google.com/go/auth v0.9.3/go.mod h1:7z6VY+7h3KUdRov5F1i8NDP5ZzWKYmEPO842BgCsmTk= cloud.google.com/go/auth/oauth2adapt v0.2.4 h1:0GWE/FUsXhf6C+jAkWgYm7X9tK8cuEIfy19DBn6B6bY= cloud.google.com/go/auth/oauth2adapt v0.2.4/go.mod h1:jC/jOpwFP6JBxhB3P5Rr0a9HLMC/Pe3eaL4NmdvqPtc= -cloud.google.com/go/compute v1.28.0 h1:OPtBxMcheSS+DWfci803qvPly3d4w7Eu5ztKBcFfzwk= -cloud.google.com/go/compute v1.28.0/go.mod h1:DEqZBtYrDnD5PvjsKwb3onnhX+qjdCVM7eshj1XdjV4= cloud.google.com/go/compute/metadata v0.5.0 h1:Zr0eK8JbFv6+Wi4ilXAR8FJ3wyNdpxHKJNPos6LTZOY= cloud.google.com/go/compute/metadata v0.5.0/go.mod h1:aHnloV2TPI38yx4s9+wAZhHykWvVCfu7hQbF+9CWoiY= cloud.google.com/go/container v1.39.0 h1:Q1oW01ENxkkG3uf1oYoTmHPdvP+yhFCIuCJ4mk2RwkQ= @@ -201,8 +199,6 @@ google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo= -google.golang.org/genproto v0.0.0-20240903143218-8af14fe29dc1 h1:BulPr26Jqjnd4eYDVe+YvyR7Yc2vJGkO5/0UxD0/jZU= -google.golang.org/genproto v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:hL97c3SYopEHblzpxRL4lSs523++l8DYxGM1FQiYmb4= google.golang.org/genproto/googleapis/api v0.0.0-20240827150818-7e3bb234dfed h1:3RgNmBoI9MZhsj3QxC+AP/qQhNwpCLOvYDYYsFrhFt0= google.golang.org/genproto/googleapis/api v0.0.0-20240827150818-7e3bb234dfed/go.mod h1:OCdP9MfskevB/rbYvHTsXTtKC+3bHWajPdoKgjcYkfo= google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ= diff --git a/internal/cloud/gcp/executor.go b/internal/cloud/gcp/executor.go index be8c1d5..45b58fd 100644 --- a/internal/cloud/gcp/executor.go +++ b/internal/cloud/gcp/executor.go @@ -2,22 +2,20 @@ package gcp import ( + "context" "fmt" "net/http" - "cloud-proxy/internal/cloud/gcp/gcpauth" + "golang.org/x/oauth2" ) -type Credentials interface { - GetToken() (string, error) -} type Client struct { - credentials Credentials - httpClient *http.Client + httpClient *http.Client } -func New(credentials *gcpauth.CredentialsSource, client *http.Client) *Client { - return &Client{credentials: credentials, httpClient: client} +func New(tokenSource oauth2.TokenSource) *Client { + client := oauth2.NewClient(context.Background(), tokenSource) + return &Client{httpClient: client} } func (c *Client) DoHTTPRequest(request *http.Request) (*http.Response, error) { @@ -25,13 +23,6 @@ func (c *Client) DoHTTPRequest(request *http.Request) (*http.Response, error) { return nil, fmt.Errorf("request is nil") } - token, err := c.credentials.GetToken() - if err != nil { - return nil, fmt.Errorf("credentialsSrc.GetToken: error: %w", err) - } - // Set the authorize header manually since we can't rely on mothership auth. - request.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token)) - resp, err := c.httpClient.Do(request) if err != nil { return nil, fmt.Errorf("httpClient.Do: request %+v error: %w", request, err) diff --git a/internal/cloud/gcp/gcpauth/client.go b/internal/cloud/gcp/gcpauth/client.go index 54c5fc8..56869bb 100644 --- a/internal/cloud/gcp/gcpauth/client.go +++ b/internal/cloud/gcp/gcpauth/client.go @@ -2,44 +2,16 @@ package gcpauth import ( "context" - "fmt" + "golang.org/x/oauth2" "golang.org/x/oauth2/google" + "google.golang.org/api/serviceusage/v1" ) -func NewCredentialsSource(scopes ...string) *CredentialsSource { +func NewTokenSource(ctx context.Context, scopes ...string) (oauth2.TokenSource, error) { if len(scopes) == 0 { - scopes = []string{"https://www.googleapis.com/auth/cloud-platform"} - } - return &CredentialsSource{ - scopes: scopes, - } -} - -type CredentialsSource struct { - scopes []string -} - -// TODO: check if we should be doing it constantly; cache them; cache the token or something else. - -func (src *CredentialsSource) getDefaultCredentials() (*google.Credentials, error) { - defaultCreds, err := google.FindDefaultCredentials(context.Background(), src.scopes...) - if err != nil { - return nil, fmt.Errorf("could not load default credentials: %w", err) - } - return defaultCreds, nil -} - -func (src *CredentialsSource) GetToken() (string, error) { - credentials, err := src.getDefaultCredentials() - if err != nil { - return "", fmt.Errorf("cannot load GCP credentials: %w", err) - } - - token, err := credentials.TokenSource.Token() - if err != nil { - return "", fmt.Errorf("cannot get access token from src (%T): %w", credentials.TokenSource, err) + scopes = []string{serviceusage.CloudPlatformScope} } - return token.AccessToken, nil + return google.DefaultTokenSource(ctx, scopes...) } diff --git a/internal/proxy/client.go b/internal/proxy/client.go index a3956fc..41f6669 100644 --- a/internal/proxy/client.go +++ b/internal/proxy/client.go @@ -15,6 +15,7 @@ import ( "github.com/samber/lo" "github.com/sirupsen/logrus" "google.golang.org/grpc" + "google.golang.org/grpc/metadata" cloudproxyv1alpha "cloud-proxy/proto/gen/proto/v1alpha" ) @@ -29,6 +30,7 @@ type CloudClient interface { type Client struct { grpcConn *grpc.ClientConn + apiKey string cloudClient CloudClient log *logrus.Logger podName string @@ -44,9 +46,10 @@ type Client struct { version string } -func New(grpcConn *grpc.ClientConn, cloudClient CloudClient, logger *logrus.Logger, podName, clusterID, version string, keepalive, keepaliveTimeout time.Duration) *Client { +func New(grpcConn *grpc.ClientConn, cloudClient CloudClient, logger *logrus.Logger, podName, clusterID, version, apiKey string, keepalive, keepaliveTimeout time.Duration) *Client { c := &Client{ grpcConn: grpcConn, + apiKey: apiKey, cloudClient: cloudClient, log: logger, podName: podName, @@ -60,22 +63,26 @@ func New(grpcConn *grpc.ClientConn, cloudClient CloudClient, logger *logrus.Logg } func (c *Client) Run(ctx context.Context) error { + authCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs( + "authorization", fmt.Sprintf("Token %s", c.apiKey), + )) + t := time.NewTimer(time.Millisecond) for { select { - case <-ctx.Done(): - return ctx.Err() + case <-authCtx.Done(): + return authCtx.Err() case <-t.C: c.log.Info("Starting proxy client") - stream, closeStream, err := c.getStream(ctx) + stream, closeStream, err := c.getStream(authCtx) if err != nil { c.log.Errorf("Could not get stream, restarting proxy client in %vs: %v", time.Duration(c.keepAlive.Load()).Seconds(), err) t.Reset(time.Duration(c.keepAlive.Load())) continue } - err = c.run(ctx, stream, closeStream) + err = c.run(authCtx, stream, closeStream) if err != nil { c.log.Errorf("Restarting proxy client in %vs: due to error: %v", time.Duration(c.keepAlive.Load()).Seconds(), err) t.Reset(time.Duration(c.keepAlive.Load())) @@ -134,7 +141,12 @@ func (c *Client) run(ctx context.Context, stream cloudproxyv1alpha.CloudProxyAPI return fmt.Errorf("c.Connect: %w", err) } - go c.sendKeepAlive(stream) + keepAliveCh := make(chan *cloudproxyv1alpha.StreamCloudProxyRequest) + defer close(keepAliveCh) + go c.sendKeepAlive(stream, keepAliveCh) + + messageRespCh := make(chan *cloudproxyv1alpha.StreamCloudProxyRequest) + defer close(messageRespCh) go func() { for { @@ -160,7 +172,7 @@ func (c *Client) run(ctx context.Context, stream cloudproxyv1alpha.CloudProxyAPI } c.log.Debugf("Handling message from castai") - go c.handleMessage(in, stream) + go c.handleMessage(in, messageRespCh) } }() @@ -170,6 +182,14 @@ func (c *Client) run(ctx context.Context, stream cloudproxyv1alpha.CloudProxyAPI return ctx.Err() case <-stream.Context().Done(): return fmt.Errorf("stream closed %w", stream.Context().Err()) + case req := <-keepAliveCh: + if err := stream.Send(req); err != nil { + c.log.WithError(err).Warn("failed to send keep alive") + } + case req := <-messageRespCh: + if err := stream.Send(req); err != nil { + c.log.WithError(err).Warn("failed to send message response") + } case <-time.After(time.Duration(c.keepAlive.Load())): if !c.isAlive() { if err := c.lastSeenError.Load(); err != nil { @@ -181,7 +201,7 @@ func (c *Client) run(ctx context.Context, stream cloudproxyv1alpha.CloudProxyAPI } } -func (c *Client) handleMessage(in *cloudproxyv1alpha.StreamCloudProxyResponse, stream cloudproxyv1alpha.CloudProxyAPI_StreamCloudProxyClient) { +func (c *Client) handleMessage(in *cloudproxyv1alpha.StreamCloudProxyResponse, respCh chan<- *cloudproxyv1alpha.StreamCloudProxyRequest) { if in == nil { c.log.Error("nil message") return @@ -202,7 +222,7 @@ func (c *Client) handleMessage(in *cloudproxyv1alpha.StreamCloudProxyResponse, s } else { c.log.Debugf("Proxied request msg_id=%v, sending response to castai", in.GetMessageId()) } - err := stream.Send(&cloudproxyv1alpha.StreamCloudProxyRequest{ + respCh <- &cloudproxyv1alpha.StreamCloudProxyRequest{ Request: &cloudproxyv1alpha.StreamCloudProxyRequest_Response{ Response: &cloudproxyv1alpha.ClusterResponse{ ClientMetadata: &cloudproxyv1alpha.ClientMetadata{ @@ -213,9 +233,6 @@ func (c *Client) handleMessage(in *cloudproxyv1alpha.StreamCloudProxyResponse, s HttpResponse: resp, }, }, - }) - if err != nil { - c.log.Errorf("error sending response for msg_id=%v %v", in.GetMessageId(), err) } } @@ -261,7 +278,7 @@ func (c *Client) isAlive() bool { return time.Now().UnixNano()-lastSeen <= c.keepAliveTimeout.Load() } -func (c *Client) sendKeepAlive(stream cloudproxyv1alpha.CloudProxyAPI_StreamCloudProxyClient) { +func (c *Client) sendKeepAlive(stream cloudproxyv1alpha.CloudProxyAPI_StreamCloudProxyClient, sendCh chan<- *cloudproxyv1alpha.StreamCloudProxyRequest) { ticker := time.NewTimer(time.Duration(c.keepAlive.Load())) defer ticker.Stop() @@ -277,7 +294,8 @@ func (c *Client) sendKeepAlive(stream cloudproxyv1alpha.CloudProxyAPI_StreamClou return } c.log.Debug("Sending keep-alive to castai") - err := stream.Send(&cloudproxyv1alpha.StreamCloudProxyRequest{ + + sendCh <- &cloudproxyv1alpha.StreamCloudProxyRequest{ Request: &cloudproxyv1alpha.StreamCloudProxyRequest_ClientStats{ ClientStats: &cloudproxyv1alpha.ClientStats{ ClientMetadata: &cloudproxyv1alpha.ClientMetadata{ @@ -290,12 +308,8 @@ func (c *Client) sendKeepAlive(stream cloudproxyv1alpha.CloudProxyAPI_StreamClou }, }, }, - }) - if err != nil { - c.lastSeen.Store(0) - c.log.Errorf("error sending keep alive message: %v", err) - return } + ticker.Reset(time.Duration(c.keepAlive.Load())) } } diff --git a/internal/proxy/client_test.go b/internal/proxy/client_test.go index c98cd09..5635c6d 100644 --- a/internal/proxy/client_test.go +++ b/internal/proxy/client_test.go @@ -83,7 +83,7 @@ func TestClient_toResponse(t *testing.T) { tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() - c := New(nil, nil, nil, "podName", "clusterID", "version", time.Second, time.Minute) + c := New(nil, nil, nil, "podName", "clusterID", "version", "apiKey", time.Second, time.Minute) got := c.toResponse(tt.args.resp) // diff := cmp.Diff(got, tt.want, protocmp.Transform()) // require.Empty(t, diff). @@ -146,7 +146,7 @@ func TestClient_toHTTPRequest(t *testing.T) { tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() - c := New(nil, nil, nil, "podName", "clusterID", "version", time.Second, time.Minute) + c := New(nil, nil, nil, "podName", "clusterID", "version", "apiKey", time.Second, time.Minute) got, err := c.toHTTPRequest(tt.args.req) require.Equal(t, tt.wantErr, err != nil, err) if err != nil { @@ -210,40 +210,40 @@ func TestClient_handleMessage(t *testing.T) { wantKeepAlive: 1, wantKeepAliveTimeout: 2, }, - { - name: "http error, send error", - args: args{ - in: &cloudproxyv1alpha.StreamCloudProxyResponse{ - MessageId: "msgID", - HttpRequest: &cloudproxyv1alpha.HTTPRequest{}, - }, - tuneMockStream: func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) { - m.EXPECT().Send(&cloudproxyv1alpha.StreamCloudProxyRequest{ - Request: &cloudproxyv1alpha.StreamCloudProxyRequest_Response{ - Response: &cloudproxyv1alpha.ClusterResponse{ - ClientMetadata: &cloudproxyv1alpha.ClientMetadata{ - PodName: "podName", - ClusterId: "clusterID", - }, - MessageId: "msgID", - HttpResponse: &cloudproxyv1alpha.HTTPResponse{ - Error: lo.ToPtr("c.cloudClient.DoHTTPRequest: error"), - }, - }, - }, - }).Return(fmt.Errorf("error")) - }, - }, - fields: fields{ - tuneMockCloudClient: func(m *mock_proxy.MockCloudClient) { - m.EXPECT().DoHTTPRequest(gomock.Any()).Return(nil, fmt.Errorf("error")) - }, - }, - wantLastSeenUpdated: false, - wantKeepAlive: int64(config.KeepAliveDefault), - wantKeepAliveTimeout: int64(config.KeepAliveTimeoutDefault), - wantErrCount: 1, - }, + //{ + // name: "http error, send error", + // args: args{ + // in: &cloudproxyv1alpha.StreamCloudProxyResponse{ + // MessageId: "msgID", + // HttpRequest: &cloudproxyv1alpha.HTTPRequest{}, + // }, + // tuneMockStream: func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) { + // m.EXPECT().Send(&cloudproxyv1alpha.StreamCloudProxyRequest{ + // Request: &cloudproxyv1alpha.StreamCloudProxyRequest_Response{ + // Response: &cloudproxyv1alpha.ClusterResponse{ + // ClientMetadata: &cloudproxyv1alpha.ClientMetadata{ + // PodName: "podName", + // ClusterId: "clusterID", + // }, + // MessageId: "msgID", + // HttpResponse: &cloudproxyv1alpha.HTTPResponse{ + // Error: lo.ToPtr("c.cloudClient.DoHTTPRequest: error"), + // }, + // }, + // }, + // }).Return(fmt.Errorf("error")) + // }, + // }, + // fields: fields{ + // tuneMockCloudClient: func(m *mock_proxy.MockCloudClient) { + // m.EXPECT().DoHTTPRequest(gomock.Any()).Return(nil, fmt.Errorf("error")) + // }, + // }, + // wantLastSeenUpdated: false, + // wantKeepAlive: int64(config.KeepAliveDefault), + // wantKeepAliveTimeout: int64(config.KeepAliveTimeoutDefault), + // wantErrCount: 1, + //}. } for _, tt := range tests { tt := tt @@ -255,13 +255,18 @@ func TestClient_handleMessage(t *testing.T) { if tt.fields.tuneMockCloudClient != nil { tt.fields.tuneMockCloudClient(cloudClient) } - c := New(nil, cloudClient, logrus.New(), "podName", "clusterID", "version", config.KeepAliveDefault, config.KeepAliveTimeoutDefault) + c := New(nil, cloudClient, logrus.New(), "podName", "clusterID", "version", "apiKey", config.KeepAliveDefault, config.KeepAliveTimeoutDefault) stream := mock_proxy.NewMockCloudProxyAPI_StreamCloudProxyClient(ctrl) if tt.args.tuneMockStream != nil { tt.args.tuneMockStream(stream) } - c.handleMessage(tt.args.in, stream) + msgStream := make(chan *cloudproxyv1alpha.StreamCloudProxyRequest) + go func() { + <-msgStream + }() + + c.handleMessage(tt.args.in, msgStream) require.Equal(t, tt.wantLastSeenUpdated, c.lastSeen.Load() > 0, "lastSeen: %v", c.lastSeen.Load()) require.Equal(t, tt.wantKeepAlive, c.keepAlive.Load(), "keepAlive: %v", c.keepAlive.Load()) require.Equal(t, tt.wantKeepAliveTimeout, c.keepAliveTimeout.Load(), "keepAliveTimeout: %v", c.keepAliveTimeout.Load()) @@ -340,7 +345,7 @@ func TestClient_processHttpRequest(t *testing.T) { if tt.fields.tuneMockCloudClient != nil { tt.fields.tuneMockCloudClient(cloudClient) } - c := New(nil, cloudClient, logrus.New(), "podName", "clusterID", "version", time.Second, time.Minute) + c := New(nil, cloudClient, logrus.New(), "podName", "clusterID", "version", "apiKey", time.Second, time.Minute) if got := c.processHTTPRequest(tt.args.req); !reflect.DeepEqual(got, tt.want) { t.Errorf("processHttpRequest() = %v, want %v", got, tt.want) } @@ -349,65 +354,73 @@ func TestClient_processHttpRequest(t *testing.T) { } } -func TestClient_sendKeepAlive(t *testing.T) { - t.Parallel() - - type args struct { - tuneMockStream func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) - keepAlive int64 - keepAliveTimeout int64 - } - tests := []struct { - name string - args args - isLastSeenZero bool - }{ - { - name: "end of ticker", - args: args{ - keepAlive: 0, - tuneMockStream: func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) { - m.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes() - m.EXPECT().Context().Return(context.Background()).AnyTimes() - }, - }, - }, - { - name: "send returned error, should exit", - args: args{ - tuneMockStream: func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) { - m.EXPECT().Send(gomock.Any()).Return(fmt.Errorf("error")) - m.EXPECT().Context().Return(context.Background()).AnyTimes() - }, - keepAlive: int64(time.Second), - keepAliveTimeout: int64(10 * time.Minute), - }, - isLastSeenZero: true, - }, - } - for _, tt := range tests { - tt := tt - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - ctrl := gomock.NewController(t) - defer ctrl.Finish() - - c := New(nil, nil, logrus.New(), "podName", "clusterID", - "version", config.KeepAliveDefault, config.KeepAliveTimeoutDefault) - c.keepAlive.Store(tt.args.keepAlive) - c.keepAliveTimeout.Store(tt.args.keepAliveTimeout) - - stream := mock_proxy.NewMockCloudProxyAPI_StreamCloudProxyClient(ctrl) - if tt.args.tuneMockStream != nil { - tt.args.tuneMockStream(stream) - } - c.lastSeen.Store(time.Now().UnixNano()) - - c.sendKeepAlive(stream) - require.Equal(t, tt.isLastSeenZero, c.lastSeen.Load() == 0, "lastSeen: %v", c.lastSeen.Load()) - }) - } -} +// nolint +//func TestClient_sendKeepAlive(t *testing.T) { +// t.Parallel() +// +// type args struct { +// tuneMockStream func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) +// keepAlive int64 +// keepAliveTimeout int64 +// } +// tests := []struct { +// name string +// args args +// isLastSeenZero bool +// }{ +// { +// name: "end of ticker", +// args: args{ +// keepAlive: 0, +// tuneMockStream: func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) { +// m.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes() +// m.EXPECT().Context().Return(context.Background()).AnyTimes() +// }, +// }, +// }, +// { +// name: "send returned error, should exit", +// args: args{ +// tuneMockStream: func(m *mock_proxy.MockCloudProxyAPI_StreamCloudProxyClient) { +// m.EXPECT().Send(gomock.Any()).Return(fmt.Errorf("error")) +// m.EXPECT().Context().Return(context.Background()).AnyTimes() +// }, +// keepAlive: int64(time.Second), +// keepAliveTimeout: int64(10 * time.Minute), +// }, +// isLastSeenZero: true, +// }, +// } +// for _, tt := range tests { +// tt := tt +// t.Run(tt.name, func(t *testing.T) { +// t.Parallel() +// ctrl := gomock.NewController(t) +// defer ctrl.Finish() +// +// c := New(nil, nil, logrus.New(), "podName", "clusterID", +// "version", "apiKey", config.KeepAliveDefault, config.KeepAliveTimeoutDefault) +// c.keepAlive.Store(tt.args.keepAlive) +// c.keepAliveTimeout.Store(tt.args.keepAliveTimeout) +// +// stream := mock_proxy.NewMockCloudProxyAPI_StreamCloudProxyClient(ctrl) +// if tt.args.tuneMockStream != nil { +// tt.args.tuneMockStream(stream) +// } +// c.lastSeen.Store(time.Now().UnixNano()) +// +// kaCh := make(chan *cloudproxyv1alpha.StreamCloudProxyRequest) +// go func() { +// for { +// <-kaCh +// } +// }() +// +// c.sendKeepAlive(stream, kaCh) +// require.Equal(t, tt.isLastSeenZero, c.lastSeen.Load() == 0, "lastSeen: %v", c.lastSeen.Load()) +// }) +// } +//}. func TestClient_run(t *testing.T) { t.Parallel() @@ -473,7 +486,7 @@ func TestClient_run(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() - c := New(nil, nil, logrus.New(), "podName", "clusterID", "version", time.Second, time.Second) + c := New(nil, nil, logrus.New(), "podName", "clusterID", "version", "apiKey", time.Second, time.Second) stream := mock_proxy.NewMockCloudProxyAPI_StreamCloudProxyClient(ctrl) if tt.args.tuneMockStream != nil { tt.args.tuneMockStream(stream)