Add client factory support for v1 CRD for Kafka and SchemaRegistry API Clients (#1074)

* Add client factory support for v1 CRD for Kafka and SchemaRegistry API clients

* DRY up pod-based URL construction and auth for listeners

Author: andrewstucki

operator/api/vectorized/v1alpha1/cluster_types.go

@@ -1330,6 +1330,20 @@ func (r *Cluster) SchemaRegistryAPITLS() *SchemaRegistryAPI {
 	return nil
 }
 
+// SchemaRegistryInternalListener returns internal listener.
+func (r *Cluster) SchemaRegistryInternalListener() *SchemaRegistryAPI {
+	if r == nil {
+		return nil
+	}
+	for i := range r.Spec.Configuration.SchemaRegistryAPI {
+		el := &r.Spec.Configuration.SchemaRegistryAPI[i]
+		if el.External == nil || !el.External.Enabled {
+			return el
+		}
+	}
+	return nil
+}
+
 // SchemaRegistryListeners returns all schema registry listeners
 func (r *Cluster) SchemaRegistryListeners() []SchemaRegistryAPI {
 	if r == nil || r.Spec.Configuration.SchemaRegistry == nil {

operator/pkg/client/cluster.go

@@ -96,6 +96,29 @@ func (c *Factory) schemaRegistryForCluster(cluster *redpandav1alpha2.Redpanda) (
 	return client, nil
 }
 
+func (c *Factory) schemaRegistryForV1Cluster(cluster *vectorizedv1alpha1.Cluster) (*sr.Client, error) {
+	ctx := context.Background()
+
+	fqdn, certs, err := v1ClusterCerts(ctx, c.Client, cluster)
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := newNodePoolInternalSchemaRegistryAPI(ctx, c.Client, cluster, fqdn, certs, c.dialer, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if c.userAuth != nil {
+		client, err = sr.NewClient(append(client.Opts(), sr.BasicAuth(c.userAuth.Username, c.userAuth.Password))...)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return client, nil
+}
+
 // kafkaForCluster returns a simple kgo.Client able to communicate with the given cluster specified via a Redpanda cluster.
 func (c *Factory) kafkaForCluster(cluster *redpandav1alpha2.Redpanda, opts ...kgo.Opt) (*kgo.Client, error) {
 	dot, err := cluster.GetDot(c.config)
@@ -129,3 +152,34 @@ func (c *Factory) kafkaForCluster(cluster *redpandav1alpha2.Redpanda, opts ...kg
 
 	return client, nil
 }
+
+func (c *Factory) kafkaForV1Cluster(cluster *vectorizedv1alpha1.Cluster, opts ...kgo.Opt) (*kgo.Client, error) {
+	ctx := context.Background()
+
+	fqdn, certs, err := v1ClusterCerts(ctx, c.Client, cluster)
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := newNodePoolInternalKafkaAPI(ctx, c.Client, cluster, fqdn, certs, c.dialer, opts)
+	if err != nil {
+		return nil, err
+	}
+
+	authOpt, err := c.kafkaUserAuth()
+	if err != nil {
+		// close the client since it's no longer usable
+		client.Close()
+
+		return nil, err
+	}
+
+	if authOpt != nil {
+		// close this client since we're not going to use it anymore
+		client.Close()
+
+		return kgo.NewClient(append(client.Opts(), authOpt)...)
+	}
+
+	return client, nil
+}

operator/pkg/client/factory.go

@@ -170,6 +170,10 @@ func (c *Factory) KafkaClient(ctx context.Context, obj any, opts ...kgo.Opt) (*k
 		return c.kafkaForCluster(cluster, opts...)
 	}
 
+	if cluster, ok := obj.(*vectorizedv1alpha1.Cluster); ok {
+		return c.kafkaForV1Cluster(cluster)
+	}
+
 	if profile, ok := obj.(*rpkconfig.RpkProfile); ok {
 		return c.kafkaForRPKProfile(profile, opts...)
 	}
@@ -236,6 +240,10 @@ func (c *Factory) SchemaRegistryClient(ctx context.Context, obj any) (*sr.Client
 		return c.schemaRegistryForCluster(cluster)
 	}
 
+	if cluster, ok := obj.(*vectorizedv1alpha1.Cluster); ok {
+		return c.schemaRegistryForV1Cluster(cluster)
+	}
+
 	if profile, ok := obj.(*rpkconfig.RpkProfile); ok {
 		return c.schemaRegistryForRPKProfile(profile)
 	}

operator/pkg/client/factory_test.go

@@ -174,6 +174,11 @@ func TestIntegrationFactoryOperatorV1(t *testing.T) {
 						Port: 9644,
 					},
 				},
+				SchemaRegistryAPI: []vectorizedv1alpha1.SchemaRegistryAPI{
+					{
+						Port: 8081,
+					},
+				},
 				DeveloperMode: true,
 			},
 		},
@@ -192,13 +197,27 @@ func TestIntegrationFactoryOperatorV1(t *testing.T) {
 		return cluster.Status.GetConditionStatus(vectorizedv1alpha1.OperatorQuiescentConditionType) == corev1.ConditionTrue
 	}, time.Minute*5, time.Second*5, "didn't work")
 
+	// check admin api
 	adminClient, err := clientFactory.RedpandaAdminClient(testutil.Context(t), &cr)
 	require.NoError(t, err)
-
 	brokers, err := adminClient.Brokers(context.Background())
 	require.NoError(t, err)
 	require.Len(t, brokers, 1)
 	require.Equal(t, rpadmin.MembershipStatusActive, brokers[0].MembershipStatus)
+
+	// check kafka api
+	kafkaClient, err := clientFactory.KafkaClient(testutil.Context(t), &cr)
+	require.NoError(t, err)
+	defer kafkaClient.Close()
+	metadata, err := kadm.NewClient(kafkaClient).BrokerMetadata(context.Background())
+	require.NoError(t, err)
+	require.Len(t, metadata.Brokers.NodeIDs(), 1)
+
+	// check schema registry api
+	srClient, err := clientFactory.SchemaRegistryClient(testutil.Context(t), &cr)
+	require.NoError(t, err)
+	_, err = srClient.SupportedTypes(context.Background())
+	require.NoError(t, err)
 }
 
 func TestIntegrationClientFactory(t *testing.T) {