diff --git a/pkg/operator/client.go b/pkg/operator/client.go index a465554..70f8a4c 100644 --- a/pkg/operator/client.go +++ b/pkg/operator/client.go @@ -14,22 +14,22 @@ import ( ) func newClient(ctx context.Context, address string) (operatorv1.RusiOperatorClient, error) { - //var retryPolicy = `{ - // "methodConfig": [{ - // "name": [{"service": "rusi.proto.operator.v1.RusiOperator"}], - // "waitForReady": true, - // "retryPolicy": { - // "MaxAttempts": 4, - // "InitialBackoff": ".01s", - // "MaxBackoff": ".01s", - // "BackoffMultiplier": 1.0, - // "RetryableStatusCodes": [ "UNAVAILABLE" ] - // } - // }]}` + var retryPolicy = `{ + "methodConfig": [{ + "name": [{"service": "rusi.proto.operator.v1.RusiOperator"}], + "waitForReady": true, + "retryPolicy": { + "MaxAttempts": 4, + "InitialBackoff": ".01s", + "MaxBackoff": ".01s", + "BackoffMultiplier": 1.0, + "RetryableStatusCodes": [ "UNAVAILABLE" ] + } + }]}` - //conn, err = grpc.DialContext(ctx, address, grpc.WithInsecure(), grpc.WithDefaultServiceConfig(retryPolicy)) - - conn, conErr := grpc.DialContext(ctx, address, grpc.WithTransportCredentials(insecure.NewCredentials())) + conn, conErr := grpc.DialContext(ctx, address, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithDefaultServiceConfig(retryPolicy)) if conErr != nil { return nil, conErr } diff --git a/pkg/operator/server.go b/pkg/operator/server.go index c29508b..ad52b7e 100644 --- a/pkg/operator/server.go +++ b/pkg/operator/server.go @@ -37,15 +37,10 @@ func (opsrv *operatorServer) WatchConfiguration(request *operatorv1.WatchConfigu } func (opsrv *operatorServer) WatchComponents(request *operatorv1.WatchComponentsRequest, stream operatorv1.RusiOperator_WatchComponentsServer) error { - - klog.V(4).InfoS("Starting streaming components") - c := make(chan rusiv1.Component) opsrv.ow.addComponentListener(c) defer opsrv.ow.removeComponentListener(c) - klog.V(4).InfoS("Starting consuming components channel") - for { select { case obj := <-c: