package cli import ( "encoding/json" "fmt" "io" "strconv" "strings" "text/tabwriter" "github.com/spf13/cobra" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" ) // queueSummary is the subset of a Queue CRD's spec/status this CLI surfaces. // Read directly off the unstructured object rather than a generated // clientset, since kmsvc-cli otherwise has no dependency on the // kafaka-management-service module's API types. type queueSummary struct { Name string `json:"name"` Namespace string `json:"namespace"` Phase string `json:"phase"` FIFO bool `json:"fifoQueue"` ShardCount int `json:"shardCount"` MaxReceives int64 `json:"maxReceiveCount"` } func newQueueCmd(flags *globalFlags) *cobra.Command { cmd := &cobra.Command{ Use: "queue", Short: "List and describe Queue CRDs (the queue-operator's source of truth)", } cmd.AddCommand( newQueueListCmd(flags), newQueueDescribeCmd(flags), ) return cmd } // queueGVK is the apiVersion/kind pair for the Queue CRD, matching queueGVR // (kmsvc.io/v1, plural "queues") by standard k8s singular-Kind convention. const ( queueAPIVersion = "kmsvc.io/v1" queueKind = "Queue" ) func newQueueCreateCmd(flags *globalFlags) *cobra.Command { var namespace string var setFields []string cmd := &cobra.Command{ Use: "create-queue [name]", Short: "Create a Queue CRD (operator defaults apply unless overridden with --set)", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { cl, err := newDynamicClient() if err != nil { return err } spec := map[string]any{} for _, kv := range setFields { if err := applySetField(spec, kv); err != nil { return err } } obj := &unstructured.Unstructured{Object: map[string]any{ "apiVersion": queueAPIVersion, "kind": queueKind, "metadata": map[string]any{ "name": args[0], "namespace": namespace, }, "spec": spec, }} created, err := cl.Resource(queueGVR).Namespace(namespace).Create(cmd.Context(), obj, metav1.CreateOptions{}) if err != nil { return fmt.Errorf("create queue %s: %w", args[0], err) } fmt.Fprintf(cmd.OutOrStdout(), "queue/%s created\n", created.GetName()) return nil }, } cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace to create the Queue CRD in") cmd.Flags().StringArrayVar(&setFields, "set", nil, "override a spec field, key=value (e.g. --set fifoQueue=true), repeatable") return cmd } func newQueueDeleteCmd(flags *globalFlags) *cobra.Command { var namespace string cmd := &cobra.Command{ Use: "delete-queue [name]", Short: "Delete a Queue CRD", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { cl, err := newDynamicClient() if err != nil { return err } if err := cl.Resource(queueGVR).Namespace(namespace).Delete(cmd.Context(), args[0], metav1.DeleteOptions{}); err != nil { return fmt.Errorf("delete queue %s: %w", args[0], err) } fmt.Fprintf(cmd.OutOrStdout(), "queue/%s deleted\n", args[0]) return nil }, } cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace the Queue CRD lives in") return cmd } // applySetField parses a "key=value" pair and writes it into spec, coercing // value to bool/int64 when it parses as one, else leaving it as a string — // mirrors helm --set's pragmatic type inference since the CRD schema isn't // known to this CLI. func applySetField(spec map[string]any, kv string) error { key, value, ok := strings.Cut(kv, "=") if !ok { return fmt.Errorf("--set %q: expected key=value", kv) } if b, err := strconv.ParseBool(value); err == nil { spec[key] = b return nil } if i, err := strconv.ParseInt(value, 10, 64); err == nil { spec[key] = i return nil } spec[key] = value return nil } func newQueueListCmd(flags *globalFlags) *cobra.Command { var namespace string cmd := &cobra.Command{ Use: "list", Short: "List Queue CRDs in a namespace", RunE: func(cmd *cobra.Command, args []string) error { cl, err := newDynamicClient() if err != nil { return err } list, err := cl.Resource(queueGVR).Namespace(namespace).List(cmd.Context(), metav1.ListOptions{}) if err != nil { return fmt.Errorf("list queues: %w", err) } summaries := make([]queueSummary, 0, len(list.Items)) for _, item := range list.Items { summaries = append(summaries, summarizeQueue(item.Object)) } return renderQueues(cmd.OutOrStdout(), flags.output, summaries) }, } cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace the Queue CRDs live in") return cmd } func newQueueDescribeCmd(flags *globalFlags) *cobra.Command { var namespace string cmd := &cobra.Command{ Use: "describe [name]", Short: "Show full status (shards, phase) for one Queue CRD", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { cl, err := newDynamicClient() if err != nil { return err } obj, err := cl.Resource(queueGVR).Namespace(namespace).Get(cmd.Context(), args[0], metav1.GetOptions{}) if err != nil { return fmt.Errorf("get queue %s: %w", args[0], err) } data, err := json.MarshalIndent(obj.Object, "", " ") if err != nil { return err } fmt.Fprintln(cmd.OutOrStdout(), string(data)) return nil }, } cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace the Queue CRD lives in") return cmd } func summarizeQueue(obj map[string]any) queueSummary { name, _, _ := unstructured.NestedString(obj, "metadata", "name") namespace, _, _ := unstructured.NestedString(obj, "metadata", "namespace") phase, _, _ := unstructured.NestedString(obj, "status", "phase") fifo, _, _ := unstructured.NestedBool(obj, "spec", "fifoQueue") maxReceives, _, _ := unstructured.NestedInt64(obj, "spec", "maxReceiveCount") shardCount := 0 if shards, ok, _ := unstructured.NestedSlice(obj, "status", "shards"); ok { shardCount = len(shards) } return queueSummary{ Name: name, Namespace: namespace, Phase: phase, FIFO: fifo, ShardCount: shardCount, MaxReceives: maxReceives, } } func renderQueues(w io.Writer, format string, queues []queueSummary) error { if format == "json" { return json.NewEncoder(w).Encode(queues) } tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0) fmt.Fprintln(tw, "NAME\tNAMESPACE\tPHASE\tFIFO\tSHARDS\tMAX_RECEIVES") for _, q := range queues { fmt.Fprintf(tw, "%s\t%s\t%s\t%t\t%d\t%d\n", q.Name, q.Namespace, q.Phase, q.FIFO, q.ShardCount, q.MaxReceives) } return tw.Flush() }