🌱 add a client to support token request

pkg/cloudevents/clients/serviceaccount/client.go

@@ -0,0 +1,158 @@
+package serviceaccount
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	authenticationv1 "k8s.io/api/authentication/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/watch"
+	applyconfigurationscorev1 "k8s.io/client-go/applyconfigurations/core/v1"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/klog/v2"
+
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/clients"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/builder"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
+	cetypes "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
+)
+
+var (
+	// TokenRequestTimeout is the timeout for CreateToken requests
+	TokenRequestTimeout = 10 * time.Second
+)
+
+type ServiceAccountClient struct {
+	grpcOptions *grpc.GRPCOptions
+	clusterName string
+}
+
+var _ corev1client.ServiceAccountInterface = &ServiceAccountClient{}
+
+// NewServiceAccountClient returns a ServiceAccountInterface
+// This client only supports creating token via gRPC in cluster namespace on the hub.
+func NewServiceAccountClient(clusterName string, opt *grpc.GRPCOptions) *ServiceAccountClient {
+	return &ServiceAccountClient{
+		grpcOptions: opt,
+		clusterName: clusterName,
+	}
+}
+
+func (sa *ServiceAccountClient) Create(ctx context.Context, serviceAccount *corev1.ServiceAccount, opts metav1.CreateOptions) (*corev1.ServiceAccount, error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "create")
+}
+
+func (sa *ServiceAccountClient) Update(ctx context.Context, serviceAccount *corev1.ServiceAccount, opts metav1.UpdateOptions) (*corev1.ServiceAccount, error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "update")
+}
+
+func (sa *ServiceAccountClient) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "delete")
+}
+
+func (sa *ServiceAccountClient) DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error {
+	return errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "delete")
+}
+
+func (sa *ServiceAccountClient) Get(ctx context.Context, name string, opts metav1.GetOptions) (*corev1.ServiceAccount, error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "get")
+}
+
+func (sa *ServiceAccountClient) List(ctx context.Context, opts metav1.ListOptions) (*corev1.ServiceAccountList, error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "list")
+}
+
+func (sa *ServiceAccountClient) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "watch")
+}
+
+func (sa *ServiceAccountClient) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *corev1.ServiceAccount, err error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "patch")
+}
+
+func (sa *ServiceAccountClient) Apply(ctx context.Context, serviceAccount *applyconfigurationscorev1.ServiceAccountApplyConfiguration, opts metav1.ApplyOptions) (result *corev1.ServiceAccount, err error) {
+	return nil, errors.NewMethodNotSupported(corev1.Resource("serviceaccounts"), "apply")
+}
+
+func (sa *ServiceAccountClient) CreateToken(ctx context.Context, serviceAccountName string, tokenRequest *authenticationv1.TokenRequest, opts metav1.CreateOptions) (*authenticationv1.TokenRequest, error) {
+	if tokenRequest == nil {
+		return nil, errors.NewBadRequest("tokenRequest is nil")
+	}
+
+	tokenRequestCtx, cancel := context.WithTimeout(ctx, TokenRequestTimeout)
+	defer cancel() // Ensure client resources are cleaned up
+
+	responseChan := make(chan *authenticationv1.TokenRequest, 1)
+
+	options, err := builder.BuildCloudEventsAgentOptions(
+		sa.grpcOptions,
+		sa.clusterName,
+		sa.clusterName,
+		TokenRequestDataType,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	cloudEventsClient, err := clients.NewCloudEventAgentClient(
+		tokenRequestCtx,
+		options,
+		nil, // resync is disabled, so lister is not required
+		nil, // resync is disabled, so statusHashGetter is not required
+		&TokenRequestCodec{},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	requestID := types.UID(uuid.New().String())
+
+	// subscribe before publish to avoid missing the response
+	cloudEventsClient.Subscribe(tokenRequestCtx, func(handlerCtx context.Context, resp *authenticationv1.TokenRequest) error {
+		if resp.UID != requestID {
+			return nil
+		}
+
+		logger := klog.FromContext(handlerCtx)
+		logger.V(4).Info("response token", "requestID", resp.UID, "serviceAccountName", resp.Name)
+		responseChan <- resp
+		return nil
+	})
+
+	// Wait for subscription to complete on the server before publishing
+	select {
+	case <-cloudEventsClient.SubscribedChan():
+		// Subscription is ready
+	case <-tokenRequestCtx.Done():
+		return nil, errors.NewInternalError(tokenRequestCtx.Err())
+	}
+
+	eventType := cetypes.CloudEventsType{
+		CloudEventsDataType: TokenRequestDataType,
+		SubResource:         cetypes.SubResourceSpec,
+		Action:              cetypes.CreateRequestAction,
+	}
+
+	newTokenRequest := tokenRequest.DeepCopy()
+	newTokenRequest.UID = requestID
+	newTokenRequest.Name = serviceAccountName
+	newTokenRequest.Namespace = sa.clusterName // the serviceaccount should locate in cluster namespace on hub
+
+	logger := klog.FromContext(tokenRequestCtx)
+	logger.V(4).Info("request token", "requestID", requestID, "serviceAccountName", serviceAccountName)
+	if err := cloudEventsClient.Publish(tokenRequestCtx, eventType, newTokenRequest); err != nil {
+		return nil, errors.NewInternalError(err)
+	}
+
+	// wait until the tokenRequestResponse is received or timeout
+	select {
+	case tokenRequestResponse := <-responseChan:
+		return tokenRequestResponse, nil
+	case <-tokenRequestCtx.Done():
+		return nil, errors.NewInternalError(tokenRequestCtx.Err())
+	}
+}

pkg/cloudevents/clients/serviceaccount/client_test.go

@@ -0,0 +1,118 @@
+package serviceaccount
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+func TestNewServiceAccountClient(t *testing.T) {
+	clusterName := "test-cluster"
+	saClient := NewServiceAccountClient(clusterName, nil)
+
+	if saClient == nil {
+		t.Fatal("expected non-nil ServiceAccountClient")
+	}
+
+	if saClient.clusterName != clusterName {
+		t.Errorf("expected clusterName %s, got %s", clusterName, saClient.clusterName)
+	}
+
+	// Verify it implements the interface
+	var _ corev1client.ServiceAccountInterface = saClient
+}
+
+func TestUnsupportedMethods(t *testing.T) {
+	saClient := &ServiceAccountClient{
+		clusterName: "test-cluster",
+	}
+
+	ctx := context.Background()
+	sa := &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-sa",
+			Namespace: "test-ns",
+		},
+	}
+
+	tests := []struct {
+		name string
+		fn   func() error
+	}{
+		{
+			name: "Create",
+			fn: func() error {
+				_, err := saClient.Create(ctx, sa, metav1.CreateOptions{})
+				return err
+			},
+		},
+		{
+			name: "Update",
+			fn: func() error {
+				_, err := saClient.Update(ctx, sa, metav1.UpdateOptions{})
+				return err
+			},
+		},
+		{
+			name: "Delete",
+			fn: func() error {
+				return saClient.Delete(ctx, "test-sa", metav1.DeleteOptions{})
+			},
+		},
+		{
+			name: "DeleteCollection",
+			fn: func() error {
+				return saClient.DeleteCollection(ctx, metav1.DeleteOptions{}, metav1.ListOptions{})
+			},
+		},
+		{
+			name: "Get",
+			fn: func() error {
+				_, err := saClient.Get(ctx, "test-sa", metav1.GetOptions{})
+				return err
+			},
+		},
+		{
+			name: "List",
+			fn: func() error {
+				_, err := saClient.List(ctx, metav1.ListOptions{})
+				return err
+			},
+		},
+		{
+			name: "Watch",
+			fn: func() error {
+				_, err := saClient.Watch(ctx, metav1.ListOptions{})
+				return err
+			},
+		},
+		{
+			name: "Patch",
+			fn: func() error {
+				_, err := saClient.Patch(ctx, "test-sa", types.StrategicMergePatchType, []byte("{}"), metav1.PatchOptions{})
+				return err
+			},
+		},
+		{
+			name: "Apply",
+			fn: func() error {
+				_, err := saClient.Apply(ctx, nil, metav1.ApplyOptions{})
+				return err
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.fn()
+			if !errors.IsMethodNotSupported(err) {
+				t.Errorf("expected MethodNotSupported error, got: %v", err)
+			}
+		})
+	}
+}

pkg/cloudevents/clients/serviceaccount/codec.go

@@ -0,0 +1,61 @@
+package serviceaccount
+
+import (
+	"fmt"
+
+	authenticationv1 "k8s.io/api/authentication/v1"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
+)
+
+var TokenRequestDataType = types.CloudEventsDataType{
+	Group:    authenticationv1.GroupName,
+	Version:  "v1",
+	Resource: "tokenrequests",
+}
+
+// TokenRequestCodec is a codec to encode/decode a event/cloudevent for an agent.
+type TokenRequestCodec struct{}
+
+func NewTokenRequestCodec() *TokenRequestCodec {
+	return &TokenRequestCodec{}
+}
+
+// EventDataType always returns the event data type `authentication.k8s.io.v1.tokenrequests`.
+func (c *TokenRequestCodec) EventDataType() types.CloudEventsDataType {
+	return TokenRequestDataType
+}
+
+// Encode the event to a cloudevent
+func (c *TokenRequestCodec) Encode(source string, eventType types.CloudEventsType, tokenRequest *authenticationv1.TokenRequest) (*cloudevents.Event, error) {
+	if tokenRequest == nil {
+		return nil, fmt.Errorf("tokenRequest is nil")
+	}
+
+	if eventType.CloudEventsDataType != TokenRequestDataType {
+		return nil, fmt.Errorf("unsupported cloudevents data type %v", eventType.CloudEventsDataType)
+	}
+
+	evt := types.NewEventBuilder(source, eventType).
+		WithResourceID(string(tokenRequest.UID)).
+		WithClusterName(tokenRequest.Namespace).
+		NewEvent()
+
+	if err := evt.SetData(cloudevents.ApplicationJSON, tokenRequest); err != nil {
+		return nil, fmt.Errorf("failed to encode event to a cloudevent: %v", err)
+	}
+
+	return &evt, nil
+}
+
+// Decode a cloudevent to an event object
+func (c *TokenRequestCodec) Decode(evt *cloudevents.Event) (*authenticationv1.TokenRequest, error) {
+	tokenRequest := &authenticationv1.TokenRequest{}
+	if err := evt.DataAs(tokenRequest); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal event data: %w", err)
+	}
+
+	return tokenRequest, nil
+}