Support data source for service account, pulsar instance and pulsar cluster

cloud/data_source_pulsar_cluster.go

@@ -0,0 +1,239 @@
+package cloud
+
+import (
+	"context"
+	"fmt"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"strings"
+)
+
+func dataSourcePulsarCluster() *schema.Resource {
+	return &schema.Resource{
+		ReadContext: dataSourcePulsarClusterRead,
+		Importer: &schema.ResourceImporter{
+			StateContext: func(ctx context.Context, d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) {
+				organizationCluster := strings.Split(d.Id(), "/")
+				_ = d.Set("organization", organizationCluster[0])
+				_ = d.Set("name", organizationCluster[1])
+				err := resourcePulsarClusterRead(ctx, d, meta)
+				if err.HasError() {
+					return nil, fmt.Errorf("import %q: %s", d.Id(), err[0].Summary)
+				}
+				return []*schema.ResourceData{d}, nil
+			},
+		},
+		Schema: map[string]*schema.Schema{
+			"organization": {
+				Type:         schema.TypeString,
+				Required:     true,
+				Description:  descriptions["organization"],
+				ValidateFunc: validateNotBlank,
+			},
+			"name": {
+				Type:         schema.TypeString,
+				Required:     true,
+				Description:  descriptions["name"],
+				ValidateFunc: validateNotBlank,
+			},
+			"instance_name": {
+				Type:        schema.TypeString,
+				Description: descriptions["instance_name"],
+				Computed:    true,
+			},
+			"location": {
+				Type:        schema.TypeString,
+				Description: descriptions["location"],
+				Computed:    true,
+			},
+			"bookie_replicas": {
+				Type:        schema.TypeInt,
+				Description: descriptions["bookie_replicas"],
+				Computed:    true,
+			},
+			"broker_replicas": {
+				Type:        schema.TypeInt,
+				Description: descriptions["broker_replicas"],
+				Computed:    true,
+			},
+			"compute_unit": {
+				Type:        schema.TypeFloat,
+				Description: descriptions["compute_unit"],
+				Computed:    true,
+			},
+			"storage_unit": {
+				Type:        schema.TypeFloat,
+				Description: descriptions["storage_unit"],
+				Computed:    true,
+			},
+			"config": {
+				Type:     schema.TypeList,
+				Optional: true,
+				Computed: true,
+				MinItems: 0,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"websocket_enabled": {
+							Type:     schema.TypeBool,
+							Optional: true,
+							Computed: true,
+						},
+						"function_enabled": {
+							Type:        schema.TypeBool,
+							Optional:    true,
+							Computed:    true,
+							Description: descriptions["function_enabled"],
+						},
+						"transaction_enabled": {
+							Type:        schema.TypeBool,
+							Optional:    true,
+							Computed:    true,
+							Description: descriptions["transaction_enabled"],
+						},
+						"protocols": {
+							Type:        schema.TypeList,
+							Optional:    true,
+							Description: descriptions["protocols"],
+							Elem: &schema.Resource{
+								Schema: map[string]*schema.Schema{
+									"kafka": {
+										Type:        schema.TypeMap,
+										Optional:    true,
+										Computed:    true,
+										Description: descriptions["kafka"],
+									},
+									"mqtt": {
+										Type:        schema.TypeMap,
+										Optional:    true,
+										Computed:    true,
+										Description: descriptions["mqtt"],
+									},
+								},
+							},
+						},
+						"audit_log": {
+							Type:        schema.TypeList,
+							Optional:    true,
+							Computed:    true,
+							Description: descriptions["audit_log"],
+							Elem: &schema.Resource{
+								Schema: map[string]*schema.Schema{
+									"categories": {
+										Type:        schema.TypeList,
+										Optional:    true,
+										Computed:    true,
+										MinItems:    1,
+										Description: descriptions["categories"],
+										Elem: &schema.Schema{
+											Type:         schema.TypeString,
+											ValidateFunc: validateAuditLog,
+										},
+									},
+								},
+							},
+						},
+						"custom": {
+							Type:        schema.TypeMap,
+							Optional:    true,
+							Computed:    true,
+							Description: descriptions["custom"],
+						},
+					},
+				},
+			},
+			"ready": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["cluster_ready"],
+			},
+			"http_tls_service_url": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["http_tls_service_url"],
+			},
+			"pulsar_tls_service_url": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["pulsar_tls_service_url"],
+			},
+			"kafka_service_url": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["kafka_service_url"],
+			},
+			"mqtt_service_url": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["mqtt_service_url"],
+			},
+			"websocket_service_url": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["websocket_service_url"],
+			},
+			"pulsar_version": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["pulsar_version"],
+			},
+			"bookkeeper_version": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["bookkeeper_version"],
+			},
+		},
+	}
+}
+
+func dataSourcePulsarClusterRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	namespace := d.Get("organization").(string)
+	name := d.Get("name").(string)
+	clientSet, err := getClientSet(getFactoryFromMeta(meta))
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("ERROR_INIT_CLIENT_ON_READ_PULSAR_CLUSTER: %w", err))
+	}
+	pulsarCluster, err := clientSet.CloudV1alpha1().PulsarClusters(namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("ERROR_READ_PULSAR_CLUSTER: %w", err))
+	}
+	_ = d.Set("ready", "False")
+	if pulsarCluster.Status.Conditions != nil {
+		for _, condition := range pulsarCluster.Status.Conditions {
+			if condition.Type == "Ready" {
+				_ = d.Set("ready", condition.Status)
+			}
+		}
+	}
+	if len(pulsarCluster.Spec.ServiceEndpoints) > 0 {
+		dnsName := pulsarCluster.Spec.ServiceEndpoints[0].DnsName
+		_ = d.Set("http_tls_service_url", fmt.Sprintf("https://%s", dnsName))
+		_ = d.Set("pulsar_tls_service_url", fmt.Sprintf("pulsar+ssl://%s:6651", dnsName))
+		if pulsarCluster.Spec.Config != nil {
+			if pulsarCluster.Spec.Config.WebsocketEnabled != nil &&
+				*pulsarCluster.Spec.Config.WebsocketEnabled {
+				_ = d.Set("websocket_service_url", fmt.Sprintf("wss://%s:9443", dnsName))
+			}
+			if pulsarCluster.Spec.Config.Protocols != nil {
+				if pulsarCluster.Spec.Config.Protocols.Kafka != nil {
+					_ = d.Set("kafka_service_url", fmt.Sprintf("%s:9093", dnsName))
+				}
+				if pulsarCluster.Spec.Config.Protocols.Mqtt != nil {
+					_ = d.Set("mqtt_service_url", fmt.Sprintf("mqtts://%s:8883", dnsName))
+				}
+			}
+		}
+	}
+	if pulsarCluster.Spec.Config != nil {
+		err = d.Set("config", flattenPulsarClusterConfig(pulsarCluster.Spec.Config))
+		if err != nil {
+			return diag.FromErr(fmt.Errorf("ERROR_READ_PULSAR_CLUSTER_CONFIG: %w", err))
+		}
+	}
+	brokerImage := strings.Split(pulsarCluster.Spec.Broker.Image, ":")
+	_ = d.Set("pulsar_version", brokerImage[1])
+	bookkeeperImage := strings.Split(pulsarCluster.Spec.BookKeeper.Image, ":")
+	_ = d.Set("bookkeeper_version", bookkeeperImage[1])
+	d.SetId(fmt.Sprintf("%s/%s", pulsarCluster.Namespace, pulsarCluster.Name))
+	return nil
+}

cloud/data_source_pulsar_instance.go

@@ -0,0 +1,93 @@
+package cloud
+
+import (
+	"context"
+	"fmt"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"strings"
+)
+
+func dataSourcePulsarInstance() *schema.Resource {
+	return &schema.Resource{
+		ReadContext: dataSourcePulsarInstanceRead,
+		Importer: &schema.ResourceImporter{
+			StateContext: func(ctx context.Context, d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) {
+				organizationInstance := strings.Split(d.Id(), "/")
+				_ = d.Set("organization", organizationInstance[0])
+				_ = d.Set("name", organizationInstance[1])
+				err := resourcePulsarInstanceRead(ctx, d, meta)
+				if err.HasError() {
+					return nil, fmt.Errorf("import %q: %s", d.Id(), err[0].Summary)
+				}
+				return []*schema.ResourceData{d}, nil
+			},
+		},
+		Schema: map[string]*schema.Schema{
+			"organization": {
+				Type:         schema.TypeString,
+				Required:     true,
+				Description:  descriptions["organization"],
+				ValidateFunc: validateNotBlank,
+			},
+			"name": {
+				Type:         schema.TypeString,
+				Required:     true,
+				Description:  descriptions["name"],
+				ValidateFunc: validateNotBlank,
+			},
+			"availability_mode": {
+				Type:     schema.TypeString,
+				Computed: true,
+				Elem: &schema.Schema{
+					Type: schema.TypeString,
+				},
+				Description: descriptions["availability-mode"],
+			},
+			"pool_name": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["pool_name"],
+			},
+			"pool_namespace": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["pool_namespace"],
+			},
+			"ready": {
+				Type:        schema.TypeString,
+				Computed:    true,
+				Description: descriptions["instance_ready"],
+			},
+		},
+	}
+}
+
+func dataSourcePulsarInstanceRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	namespace := d.Get("organization").(string)
+	name := d.Get("name").(string)
+	clientSet, err := getClientSet(getFactoryFromMeta(meta))
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("ERROR_INIT_CLIENT_ON_READ_SERVICE_ACCOUNT: %w", err))
+	}
+	pulsarInstance, err := clientSet.CloudV1alpha1().PulsarInstances(namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("ERROR_READ_PULSAR_INSTANCE: %w", err))
+	}
+	_ = d.Set("ready", "False")
+	if pulsarInstance.Status.Conditions != nil {
+		for _, condition := range pulsarInstance.Status.Conditions {
+			if condition.Type == "Ready" && condition.Status == "True" {
+				_ = d.Set("ready", "True")
+			}
+		}
+	}
+	if pulsarInstance.Spec.PoolRef != nil {
+		_ = d.Set("pool_name", pulsarInstance.Spec.PoolRef.Name)
+		_ = d.Set("pool_namespace", pulsarInstance.Spec.PoolRef.Namespace)
+	}
+	_ = d.Set("availability_mode", pulsarInstance.Spec.AvailabilityMode)
+	d.SetId(fmt.Sprintf("%s/%s", pulsarInstance.Namespace, pulsarInstance.Name))
+	return nil
+}