(chore) setup kmsvc-cli
This commit is contained in:
Story Crater Bot
@@ -0,0 +1,100 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
||||
kmsvc "forgejo.riotpiao.homelab.com/homelab/kmsvc-sdk"
|
||||
"google.golang.org/grpc/credentials"
|
||||
)
|
||||
|
||||
// defaultTokenURL is the Authentik OAuth2 token endpoint used with
|
||||
// --client-id/--client-secret when --token-url/KMSVC_TOKEN_URL is unset.
|
||||
const defaultTokenURL = "https://authentik.riotpiao.homelab.com/application/o/token/"
|
||||
|
||||
// buildClient constructs a *kmsvc.Client from resolved global flags.
|
||||
//
|
||||
// Defaults to TLS: the SDK itself defaults to plaintext (appropriate for
|
||||
// cluster-internal callers), but kmsvc-cli's own default --server
|
||||
// (kmsvc.riotpiao.homelab.com:443, see README) is reached through an
|
||||
// Ingress-terminated HTTPS/gRPC-passthrough endpoint, so a real external
|
||||
// invocation needs a TLS handshake, not plaintext. --insecure opts back into
|
||||
// plaintext for cluster-internal/dev targets.
|
||||
func buildClient(ctx context.Context, flags *globalFlags) (*kmsvc.Client, error) {
|
||||
if flags.server == "" {
|
||||
return nil, fmt.Errorf("server address required (--server or KMSVC_SERVER)")
|
||||
}
|
||||
|
||||
token := flags.token
|
||||
if token == "" && flags.clientID != "" && flags.clientSecret != "" {
|
||||
tokenURL := flags.tokenURL
|
||||
if tokenURL == "" {
|
||||
tokenURL = defaultTokenURL
|
||||
}
|
||||
fetched, err := fetchClientCredentialsToken(ctx, tokenURL, flags.clientID, flags.clientSecret)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("fetch token via client_credentials: %w", err)
|
||||
}
|
||||
token = fetched
|
||||
}
|
||||
|
||||
var opts []kmsvc.Option
|
||||
if token != "" {
|
||||
opts = append(opts, kmsvc.WithTokenSource(kmsvc.StaticToken(token)))
|
||||
}
|
||||
if !flags.insecure {
|
||||
opts = append(opts, kmsvc.WithTransportCredentials(credentials.NewTLS(nil)))
|
||||
}
|
||||
|
||||
client, err := kmsvc.New(ctx, flags.server, opts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect to %s: %w", flags.server, err)
|
||||
}
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// fetchClientCredentialsToken performs an OAuth2 client_credentials grant
|
||||
// against tokenURL, used when --token/KMSVC_TOKEN is unset but
|
||||
// --client-id/--client-secret (KMSVC_CLIENT_ID/KMSVC_CLIENT_SECRET) are
|
||||
// configured, so callers don't need a separate curl step to mint a token.
|
||||
func fetchClientCredentialsToken(ctx context.Context, tokenURL, clientID, clientSecret string) (string, error) {
|
||||
if tokenURL == "" {
|
||||
return "", fmt.Errorf("token URL required (--token-url or KMSVC_TOKEN_URL)")
|
||||
}
|
||||
|
||||
form := url.Values{
|
||||
"grant_type": {"client_credentials"},
|
||||
"client_id": {clientID},
|
||||
"client_secret": {clientSecret},
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, tokenURL, strings.NewReader(form.Encode()))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("token endpoint returned %s", resp.Status)
|
||||
}
|
||||
|
||||
var body struct {
|
||||
AccessToken string `json:"access_token"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
|
||||
return "", fmt.Errorf("decode token response: %w", err)
|
||||
}
|
||||
if body.AccessToken == "" {
|
||||
return "", fmt.Errorf("token response missing access_token")
|
||||
}
|
||||
return body.AccessToken, nil
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
// Config holds resolved CLI defaults. Precedence (highest to lowest):
|
||||
// command-line flag > environment variable > ~/.kmsvc/config.yaml > built-in default.
|
||||
type Config struct {
|
||||
Server string
|
||||
Token string
|
||||
Output string
|
||||
Insecure bool
|
||||
ClientID string
|
||||
ClientSecret string
|
||||
TokenURL string
|
||||
}
|
||||
|
||||
type fileConfig struct {
|
||||
Server string `yaml:"server"`
|
||||
Output string `yaml:"output"`
|
||||
Insecure bool `yaml:"insecure"`
|
||||
ClientID string `yaml:"clientId"`
|
||||
ClientSecret string `yaml:"clientSecret"`
|
||||
TokenURL string `yaml:"tokenUrl"`
|
||||
}
|
||||
|
||||
// LoadConfig resolves defaults from ~/.kmsvc/config.yaml and environment
|
||||
// variables (KMSVC_SERVER, KMSVC_TOKEN, KMSVC_OUTPUT, KMSVC_INSECURE,
|
||||
// KMSVC_CLIENT_ID, KMSVC_CLIENT_SECRET, KMSVC_TOKEN_URL). ClientID/ClientSecret
|
||||
// additionally fall back to AUTHENTIK_KAFAKA_CLIENT_ID/AUTHENTIK_KAFAKA_CLIENT_SECRET
|
||||
// (the homelab's Authentik app credentials, typically exported into the shell
|
||||
// via `vsource` from Vault) when the KMSVC_-prefixed vars aren't set, so this
|
||||
// CLI doesn't need its own separately-exported copy. Flags are applied on top
|
||||
// of this by the caller (root.go), so this function never reads flags.
|
||||
func LoadConfig() Config {
|
||||
cfg := Config{Output: "table"}
|
||||
|
||||
if path, err := configFilePath(); err == nil {
|
||||
if fc, err := readFileConfig(path); err == nil {
|
||||
if fc.Server != "" {
|
||||
cfg.Server = fc.Server
|
||||
}
|
||||
if fc.Output != "" {
|
||||
cfg.Output = fc.Output
|
||||
}
|
||||
cfg.Insecure = fc.Insecure
|
||||
if fc.ClientID != "" {
|
||||
cfg.ClientID = fc.ClientID
|
||||
}
|
||||
if fc.ClientSecret != "" {
|
||||
cfg.ClientSecret = fc.ClientSecret
|
||||
}
|
||||
if fc.TokenURL != "" {
|
||||
cfg.TokenURL = fc.TokenURL
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if v := os.Getenv("KMSVC_SERVER"); v != "" {
|
||||
cfg.Server = v
|
||||
}
|
||||
if v := os.Getenv("KMSVC_TOKEN"); v != "" {
|
||||
cfg.Token = v
|
||||
}
|
||||
if v := os.Getenv("KMSVC_OUTPUT"); v != "" {
|
||||
cfg.Output = v
|
||||
}
|
||||
if v := os.Getenv("KMSVC_INSECURE"); v != "" {
|
||||
cfg.Insecure = v == "1" || v == "true"
|
||||
}
|
||||
if v := os.Getenv("AUTHENTIK_KAFAKA_CLIENT_ID"); v != "" {
|
||||
cfg.ClientID = v
|
||||
}
|
||||
if v := os.Getenv("KMSVC_CLIENT_ID"); v != "" {
|
||||
cfg.ClientID = v
|
||||
}
|
||||
if v := os.Getenv("AUTHENTIK_KAFAKA_CLIENT_SECRET"); v != "" {
|
||||
cfg.ClientSecret = v
|
||||
}
|
||||
if v := os.Getenv("KMSVC_CLIENT_SECRET"); v != "" {
|
||||
cfg.ClientSecret = v
|
||||
}
|
||||
if v := os.Getenv("KMSVC_TOKEN_URL"); v != "" {
|
||||
cfg.TokenURL = v
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
||||
func configFilePath() (string, error) {
|
||||
home, err := os.UserHomeDir()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return filepath.Join(home, ".kmsvc", "config.yaml"), nil
|
||||
}
|
||||
|
||||
func readFileConfig(path string) (fileConfig, error) {
|
||||
var fc fileConfig
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return fc, err
|
||||
}
|
||||
if err := yaml.Unmarshal(data, &fc); err != nil {
|
||||
return fc, err
|
||||
}
|
||||
return fc, nil
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestLoadConfigDefaults(t *testing.T) {
|
||||
t.Setenv("HOME", t.TempDir())
|
||||
t.Setenv("KMSVC_SERVER", "")
|
||||
t.Setenv("KMSVC_TOKEN", "")
|
||||
t.Setenv("KMSVC_OUTPUT", "")
|
||||
|
||||
cfg := LoadConfig()
|
||||
if cfg.Output != "table" {
|
||||
t.Errorf("Output = %q, want default %q", cfg.Output, "table")
|
||||
}
|
||||
if cfg.Server != "" || cfg.Token != "" {
|
||||
t.Errorf("expected empty Server/Token with no file or env, got %+v", cfg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadConfigFileThenEnvOverride(t *testing.T) {
|
||||
home := t.TempDir()
|
||||
t.Setenv("HOME", home)
|
||||
|
||||
dir := filepath.Join(home, ".kmsvc")
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
content := "server: file-server:443\noutput: json\n"
|
||||
if err := os.WriteFile(filepath.Join(dir, "config.yaml"), []byte(content), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Setenv("KMSVC_SERVER", "")
|
||||
t.Setenv("KMSVC_OUTPUT", "")
|
||||
t.Setenv("KMSVC_TOKEN", "")
|
||||
|
||||
cfg := LoadConfig()
|
||||
if cfg.Server != "file-server:443" || cfg.Output != "json" {
|
||||
t.Fatalf("expected file values, got %+v", cfg)
|
||||
}
|
||||
|
||||
// Env var must win over the file.
|
||||
t.Setenv("KMSVC_SERVER", "env-server:443")
|
||||
cfg = LoadConfig()
|
||||
if cfg.Server != "env-server:443" {
|
||||
t.Errorf("Server = %q, want env override %q", cfg.Server, "env-server:443")
|
||||
}
|
||||
if cfg.Output != "json" {
|
||||
t.Errorf("Output = %q, want file value %q to survive (env unset)", cfg.Output, "json")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
kmsvc "forgejo.riotpiao.homelab.com/homelab/kmsvc-sdk"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
func newDLQCmd(flags *globalFlags) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: "dlq",
|
||||
Short: "Inspect and redrive dead-letter queues",
|
||||
}
|
||||
cmd.AddCommand(newDLQPeekCmd(flags), newDLQRedriveCmd(flags))
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newDLQPeekCmd(flags *globalFlags) *cobra.Command {
|
||||
var queue string
|
||||
var maxMessages, visibilityTimeout int32
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "peek",
|
||||
Short: "Receive messages from a DLQ without deleting them",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
client, err := buildClient(cmd.Context(), flags)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
msgs, err := client.ReceiveMessage(cmd.Context(), queue, kmsvc.ReceiveOptions{
|
||||
MaxNumberOfMessages: maxMessages,
|
||||
VisibilityTimeoutSeconds: visibilityTimeout,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return renderMessages(cmd.OutOrStdout(), flags.output, msgs)
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().StringVar(&queue, "queue", "", "DLQ name (required)")
|
||||
cmd.Flags().Int32Var(&maxMessages, "max-messages", 10, "maximum number of messages to peek (1-10)")
|
||||
cmd.Flags().Int32Var(&visibilityTimeout, "visibility-timeout", 5, "visibility timeout for the peek, in seconds — keep short so messages reappear quickly")
|
||||
cmd.MarkFlagRequired("queue")
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newDLQRedriveCmd(flags *globalFlags) *cobra.Command {
|
||||
var queue, to string
|
||||
var maxMessages int32
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "redrive",
|
||||
Short: "Move messages from a DLQ back to their source queue",
|
||||
Long: "Receives messages from --queue (the DLQ), sends each to --to (the source\n" +
|
||||
"queue), then deletes it from the DLQ. This is 3+ separate SDK calls, not an\n" +
|
||||
"atomic operation: if send succeeds but delete fails, the message is reported\n" +
|
||||
"as sent-but-not-removed (it may be redelivered from both queues), and the\n" +
|
||||
"command exits non-zero rather than silently continuing.",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
client, err := buildClient(cmd.Context(), flags)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
msgs, err := client.ReceiveMessage(cmd.Context(), queue, kmsvc.ReceiveOptions{
|
||||
MaxNumberOfMessages: maxMessages,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("receive from %s: %w", queue, err)
|
||||
}
|
||||
|
||||
out := cmd.OutOrStdout()
|
||||
var failures int
|
||||
for _, m := range msgs {
|
||||
sendOut, err := client.SendMessage(cmd.Context(), kmsvc.SendMessageInput{
|
||||
QueueName: to,
|
||||
Body: m.Body,
|
||||
})
|
||||
if err != nil {
|
||||
fmt.Fprintf(out, "redrive %s: send to %s failed, message left in DLQ: %v\n", m.MessageID, to, err)
|
||||
failures++
|
||||
continue
|
||||
}
|
||||
|
||||
if err := client.DeleteMessage(cmd.Context(), queue, m.ReceiptHandle); err != nil {
|
||||
fmt.Fprintf(out, "redrive %s: sent to %s as %s, but delete from %s failed — message may be duplicated: %v\n", m.MessageID, to, sendOut.MessageID, queue, err)
|
||||
failures++
|
||||
continue
|
||||
}
|
||||
|
||||
fmt.Fprintf(out, "redrive %s: sent to %s as %s, removed from %s\n", m.MessageID, to, sendOut.MessageID, queue)
|
||||
}
|
||||
|
||||
if failures > 0 {
|
||||
return fmt.Errorf("%d/%d message(s) failed to fully redrive", failures, len(msgs))
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().StringVar(&queue, "queue", "", "DLQ name (required)")
|
||||
cmd.Flags().StringVar(&to, "to", "", "source queue to redrive messages back to (required)")
|
||||
cmd.Flags().Int32Var(&maxMessages, "max-messages", 10, "maximum number of messages to redrive in this run (1-10)")
|
||||
cmd.MarkFlagRequired("queue")
|
||||
cmd.MarkFlagRequired("to")
|
||||
|
||||
return cmd
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
kafkamgmtv1 "forgejo.riotpiao.homelab.com/homelab/kmsvc-proto/gen/kafkamgmt/v1"
|
||||
)
|
||||
|
||||
func TestDLQRedriveHappyPath(t *testing.T) {
|
||||
var sendCalled, deleteCalled atomic.Bool
|
||||
var sendQueue string
|
||||
|
||||
fake := &fakeQueueService{
|
||||
receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) {
|
||||
return &kafkamgmtv1.ReceiveMessageResponse{
|
||||
Messages: []*kafkamgmtv1.Message{{MessageId: "m1", ReceiptHandle: "rh1", Body: []byte("payload")}},
|
||||
}, nil
|
||||
},
|
||||
sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
|
||||
sendCalled.Store(true)
|
||||
sendQueue = req.QueueName
|
||||
if deleteCalled.Load() {
|
||||
t.Error("delete called before send completed")
|
||||
}
|
||||
return &kafkamgmtv1.SendMessageResponse{MessageId: "m1-redriven"}, nil
|
||||
},
|
||||
deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
|
||||
deleteCalled.Store(true)
|
||||
if !sendCalled.Load() {
|
||||
t.Error("delete called before send")
|
||||
}
|
||||
return &kafkamgmtv1.DeleteMessageResponse{}, nil
|
||||
},
|
||||
}
|
||||
addr := startTestServer(t, fake)
|
||||
|
||||
flags := &globalFlags{server: addr, output: "table", insecure: true}
|
||||
cmd := newDLQRedriveCmd(flags)
|
||||
cmd.SetArgs([]string{"--queue", "orders.dlq", "--to", "orders"})
|
||||
|
||||
var out bytes.Buffer
|
||||
cmd.SetOut(&out)
|
||||
if err := cmd.Execute(); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
if !sendCalled.Load() || !deleteCalled.Load() {
|
||||
t.Fatal("expected both send and delete to be called")
|
||||
}
|
||||
if sendQueue != "orders" {
|
||||
t.Errorf("send queue = %q, want orders", sendQueue)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDLQRedriveSurfacesDeleteFailure(t *testing.T) {
|
||||
fake := &fakeQueueService{
|
||||
receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) {
|
||||
return &kafkamgmtv1.ReceiveMessageResponse{
|
||||
Messages: []*kafkamgmtv1.Message{{MessageId: "m1", ReceiptHandle: "rh1", Body: []byte("payload")}},
|
||||
}, nil
|
||||
},
|
||||
sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
|
||||
return &kafkamgmtv1.SendMessageResponse{MessageId: "m1-redriven"}, nil
|
||||
},
|
||||
deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
|
||||
return nil, errBoom
|
||||
},
|
||||
}
|
||||
addr := startTestServer(t, fake)
|
||||
|
||||
flags := &globalFlags{server: addr, output: "table", insecure: true}
|
||||
cmd := newDLQRedriveCmd(flags)
|
||||
cmd.SetArgs([]string{"--queue", "orders.dlq", "--to", "orders"})
|
||||
cmd.SilenceUsage = true
|
||||
cmd.SilenceErrors = true
|
||||
|
||||
var out bytes.Buffer
|
||||
cmd.SetOut(&out)
|
||||
|
||||
err := cmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("expected redrive to report failure when delete fails")
|
||||
}
|
||||
|
||||
output := out.String()
|
||||
if !strings.Contains(output, "may be duplicated") {
|
||||
t.Errorf("output = %q, want a duplicate-risk warning", output)
|
||||
}
|
||||
}
|
||||
|
||||
var errBoom = &boomError{}
|
||||
|
||||
type boomError struct{}
|
||||
|
||||
func (e *boomError) Error() string { return "boom" }
|
||||
@@ -0,0 +1,42 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/tools/clientcmd"
|
||||
)
|
||||
|
||||
// queueGVR identifies the Queue CRD (kmsvc.io/v1, plural "queues") that the
|
||||
// queue-operator reconciles -- the declarative source of truth for queues,
|
||||
// not something management-service's gRPC API manages (design.md §2a).
|
||||
var queueGVR = schema.GroupVersionResource{Group: "kmsvc.io", Version: "v1", Resource: "queues"}
|
||||
|
||||
// newDynamicClient builds a k8s dynamic client from the default kubeconfig
|
||||
// (KUBECONFIG env var, falling back to ~/.kube/config), the same resolution
|
||||
// kubectl itself uses.
|
||||
func newDynamicClient() (dynamic.Interface, error) {
|
||||
path, err := kubeconfigPath()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cfg, err := clientcmd.BuildConfigFromFlags("", path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load kubeconfig %s: %w", path, err)
|
||||
}
|
||||
return dynamic.NewForConfig(cfg)
|
||||
}
|
||||
|
||||
func kubeconfigPath() (string, error) {
|
||||
if v := os.Getenv("KUBECONFIG"); v != "" {
|
||||
return v, nil
|
||||
}
|
||||
home, err := os.UserHomeDir()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return filepath.Join(home, ".kube", "config"), nil
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
kmsvc "forgejo.riotpiao.homelab.com/homelab/kmsvc-sdk"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
func newMessageSendCmd(flags *globalFlags) *cobra.Command {
|
||||
var queue, body, groupID, dedupID string
|
||||
var delaySeconds int32
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "send-message",
|
||||
Short: "Send a message to a queue",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
client, err := buildClient(cmd.Context(), flags)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
out, err := client.SendMessage(cmd.Context(), kmsvc.SendMessageInput{
|
||||
QueueName: queue,
|
||||
Body: []byte(body),
|
||||
MessageGroupID: groupID,
|
||||
MessageDeduplicationID: dedupID,
|
||||
DelaySeconds: delaySeconds,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fmt.Fprintf(cmd.OutOrStdout(), "message_id=%s sequence_number=%s\n", out.MessageID, out.SequenceNumber)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
||||
cmd.Flags().StringVar(&body, "body", "", "message body (required)")
|
||||
cmd.Flags().StringVar(&groupID, "group-id", "", "FIFO message group ID")
|
||||
cmd.Flags().StringVar(&dedupID, "dedup-id", "", "FIFO message deduplication ID")
|
||||
cmd.Flags().Int32Var(&delaySeconds, "delay", 0, "delay before the message becomes visible, in seconds")
|
||||
cmd.MarkFlagRequired("queue")
|
||||
cmd.MarkFlagRequired("body")
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newMessageReceiveCmd(flags *globalFlags) *cobra.Command {
|
||||
var queue string
|
||||
var maxMessages, waitSeconds, visibilityTimeout int32
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "receive-message",
|
||||
Short: "Receive messages from a queue (long-poll)",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
client, err := buildClient(cmd.Context(), flags)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
msgs, err := client.ReceiveMessage(cmd.Context(), queue, kmsvc.ReceiveOptions{
|
||||
MaxNumberOfMessages: maxMessages,
|
||||
WaitTimeSeconds: waitSeconds,
|
||||
VisibilityTimeoutSeconds: visibilityTimeout,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return renderMessages(cmd.OutOrStdout(), flags.output, msgs)
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
||||
cmd.Flags().Int32Var(&maxMessages, "max-messages", 1, "maximum number of messages to return (1-10)")
|
||||
cmd.Flags().Int32Var(&waitSeconds, "wait", 0, "long-poll wait time in seconds (0-20)")
|
||||
cmd.Flags().Int32Var(&visibilityTimeout, "visibility-timeout", 0, "override the queue's default visibility timeout, in seconds")
|
||||
cmd.MarkFlagRequired("queue")
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newMessageDeleteCmd(flags *globalFlags) *cobra.Command {
|
||||
var queue, receiptHandle string
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "delete-message",
|
||||
Short: "Delete (acknowledge) a message",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
client, err := buildClient(cmd.Context(), flags)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
if err := client.DeleteMessage(cmd.Context(), queue, receiptHandle); err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Fprintln(cmd.OutOrStdout(), "deleted")
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
||||
cmd.Flags().StringVar(&receiptHandle, "receipt-handle", "", "receipt handle from receive (required)")
|
||||
cmd.MarkFlagRequired("queue")
|
||||
cmd.MarkFlagRequired("receipt-handle")
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newMessageChangeVisibilityCmd(flags *globalFlags) *cobra.Command {
|
||||
var queue, receiptHandle string
|
||||
var timeout int32
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "change-message-visibility",
|
||||
Short: "Change the visibility timeout of an in-flight message",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
client, err := buildClient(cmd.Context(), flags)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
if err := client.ChangeMessageVisibility(cmd.Context(), queue, receiptHandle, timeout); err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Fprintln(cmd.OutOrStdout(), "updated")
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
||||
cmd.Flags().StringVar(&receiptHandle, "receipt-handle", "", "receipt handle from receive (required)")
|
||||
cmd.Flags().Int32Var(&timeout, "timeout", 0, "new visibility timeout, in seconds (required)")
|
||||
cmd.MarkFlagRequired("queue")
|
||||
cmd.MarkFlagRequired("receipt-handle")
|
||||
cmd.MarkFlagRequired("timeout")
|
||||
|
||||
return cmd
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
kafkamgmtv1 "forgejo.riotpiao.homelab.com/homelab/kmsvc-proto/gen/kafkamgmt/v1"
|
||||
)
|
||||
|
||||
func TestMessageSendCmd(t *testing.T) {
|
||||
fake := &fakeQueueService{
|
||||
sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
|
||||
if req.QueueName != "orders" || string(req.MessageBody) != "hello" {
|
||||
t.Errorf("unexpected request: %+v", req)
|
||||
}
|
||||
return &kafkamgmtv1.SendMessageResponse{MessageId: "m1"}, nil
|
||||
},
|
||||
}
|
||||
addr := startTestServer(t, fake)
|
||||
|
||||
flags := &globalFlags{server: addr, output: "table", insecure: true}
|
||||
cmd := newMessageSendCmd(flags)
|
||||
cmd.SetArgs([]string{"--queue", "orders", "--body", "hello"})
|
||||
|
||||
var out bytes.Buffer
|
||||
cmd.SetOut(&out)
|
||||
if err := cmd.Execute(); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
|
||||
if !strings.Contains(out.String(), "message_id=m1") {
|
||||
t.Errorf("output = %q, want message_id=m1", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessageReceiveCmdJSON(t *testing.T) {
|
||||
fake := &fakeQueueService{
|
||||
receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) {
|
||||
return &kafkamgmtv1.ReceiveMessageResponse{
|
||||
Messages: []*kafkamgmtv1.Message{{MessageId: "m1", ReceiptHandle: "rh1", Body: []byte("hi")}},
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
addr := startTestServer(t, fake)
|
||||
|
||||
flags := &globalFlags{server: addr, output: "json", insecure: true}
|
||||
cmd := newMessageReceiveCmd(flags)
|
||||
cmd.SetArgs([]string{"--queue", "orders"})
|
||||
|
||||
var out bytes.Buffer
|
||||
cmd.SetOut(&out)
|
||||
if err := cmd.Execute(); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
|
||||
if !strings.Contains(out.String(), `"message_id":"m1"`) && !strings.Contains(out.String(), `"MessageID":"m1"`) {
|
||||
t.Errorf("output = %q, want JSON containing message id m1", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessageDeleteCmd(t *testing.T) {
|
||||
var gotHandle string
|
||||
fake := &fakeQueueService{
|
||||
deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
|
||||
gotHandle = req.ReceiptHandle
|
||||
return &kafkamgmtv1.DeleteMessageResponse{}, nil
|
||||
},
|
||||
}
|
||||
addr := startTestServer(t, fake)
|
||||
|
||||
flags := &globalFlags{server: addr, output: "table", insecure: true}
|
||||
cmd := newMessageDeleteCmd(flags)
|
||||
cmd.SetArgs([]string{"--queue", "orders", "--receipt-handle", "rh-1"})
|
||||
|
||||
var out bytes.Buffer
|
||||
cmd.SetOut(&out)
|
||||
if err := cmd.Execute(); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
if gotHandle != "rh-1" {
|
||||
t.Errorf("ReceiptHandle = %q, want rh-1", gotHandle)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessageSendCmdRequiresQueueAndBody(t *testing.T) {
|
||||
flags := &globalFlags{server: "unused:1", output: "table"}
|
||||
cmd := newMessageSendCmd(flags)
|
||||
cmd.SetArgs([]string{})
|
||||
cmd.SilenceUsage = true
|
||||
cmd.SilenceErrors = true
|
||||
|
||||
if err := cmd.Execute(); err == nil {
|
||||
t.Fatal("expected error for missing required flags")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"text/tabwriter"
|
||||
|
||||
kmsvc "forgejo.riotpiao.homelab.com/homelab/kmsvc-sdk"
|
||||
)
|
||||
|
||||
// renderMessages writes msgs to w in the requested format ("table" or
|
||||
// "json"). Unrecognized formats fall back to "table".
|
||||
func renderMessages(w io.Writer, format string, msgs []kmsvc.Message) error {
|
||||
if format == "json" {
|
||||
return json.NewEncoder(w).Encode(msgs)
|
||||
}
|
||||
|
||||
tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
|
||||
fmt.Fprintln(tw, "MESSAGE_ID\tRECEIPT_HANDLE\tRECEIVE_COUNT\tBODY")
|
||||
for _, m := range msgs {
|
||||
fmt.Fprintf(tw, "%s\t%s\t%d\t%s\n", m.MessageID, m.ReceiptHandle, m.ReceiveCount, string(m.Body))
|
||||
}
|
||||
return tw.Flush()
|
||||
}
|
||||
|
||||
// renderBatchResult writes a batch send/delete result to w.
|
||||
func renderBatchResult(w io.Writer, format string, successful, failed []kmsvc.BatchResultEntry) error {
|
||||
if format == "json" {
|
||||
return json.NewEncoder(w).Encode(struct {
|
||||
Successful []kmsvc.BatchResultEntry `json:"successful"`
|
||||
Failed []kmsvc.BatchResultEntry `json:"failed"`
|
||||
}{successful, failed})
|
||||
}
|
||||
|
||||
tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
|
||||
fmt.Fprintln(tw, "ID\tSTATUS\tMESSAGE_ID\tERROR")
|
||||
for _, e := range successful {
|
||||
fmt.Fprintf(tw, "%s\tok\t%s\t\n", e.ID, e.MessageID)
|
||||
}
|
||||
for _, e := range failed {
|
||||
fmt.Fprintf(tw, "%s\tfailed\t\t%s\n", e.ID, e.Error)
|
||||
}
|
||||
return tw.Flush()
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
kmsvc "forgejo.riotpiao.homelab.com/homelab/kmsvc-sdk"
|
||||
)
|
||||
|
||||
func TestRenderMessagesTable(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
msgs := []kmsvc.Message{{MessageID: "m1", ReceiptHandle: "rh1", Body: []byte("hi"), ReceiveCount: 2}}
|
||||
|
||||
if err := renderMessages(&buf, "table", msgs); err != nil {
|
||||
t.Fatalf("renderMessages: %v", err)
|
||||
}
|
||||
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "MESSAGE_ID") || !strings.Contains(out, "m1") || !strings.Contains(out, "hi") {
|
||||
t.Errorf("table output missing expected content: %q", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRenderMessagesJSON(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
msgs := []kmsvc.Message{{MessageID: "m1", ReceiptHandle: "rh1", Body: []byte("hi")}}
|
||||
|
||||
if err := renderMessages(&buf, "json", msgs); err != nil {
|
||||
t.Fatalf("renderMessages: %v", err)
|
||||
}
|
||||
|
||||
var got []kmsvc.Message
|
||||
if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
|
||||
t.Fatalf("unmarshal output: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0].MessageID != "m1" {
|
||||
t.Errorf("unexpected decoded messages: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRenderBatchResultTable(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
successful := []kmsvc.BatchResultEntry{{ID: "1", MessageID: "m1"}}
|
||||
failed := []kmsvc.BatchResultEntry{{ID: "2", Error: "boom"}}
|
||||
|
||||
if err := renderBatchResult(&buf, "table", successful, failed); err != nil {
|
||||
t.Fatalf("renderBatchResult: %v", err)
|
||||
}
|
||||
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "ok") || !strings.Contains(out, "boom") {
|
||||
t.Errorf("table output missing expected content: %q", out)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
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()
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
// globalFlags holds the resolved values of the root command's persistent
|
||||
// flags after parsing — read by subcommands when they build a client.
|
||||
type globalFlags struct {
|
||||
server string
|
||||
token string
|
||||
output string
|
||||
insecure bool
|
||||
clientID string
|
||||
clientSecret string
|
||||
tokenURL string
|
||||
}
|
||||
|
||||
// NewRootCmd builds the kmsvc root command.
|
||||
func NewRootCmd() *cobra.Command {
|
||||
cfg := LoadConfig()
|
||||
flags := &globalFlags{
|
||||
server: cfg.Server,
|
||||
output: cfg.Output,
|
||||
insecure: cfg.Insecure,
|
||||
clientID: cfg.ClientID,
|
||||
tokenURL: cfg.TokenURL,
|
||||
}
|
||||
|
||||
root := &cobra.Command{
|
||||
Use: "kmsvc",
|
||||
Short: "Kafka Management Service CLI",
|
||||
SilenceUsage: true,
|
||||
SilenceErrors: false,
|
||||
// Secrets resolved from env/config aren't pre-bound to the flag's
|
||||
// pflag default (which --help prints verbatim) — apply them here
|
||||
// instead, only when the user didn't pass the flag explicitly.
|
||||
PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
|
||||
if !cmd.Flags().Changed("token") {
|
||||
flags.token = cfg.Token
|
||||
}
|
||||
if !cmd.Flags().Changed("client-secret") {
|
||||
flags.clientSecret = cfg.ClientSecret
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
root.PersistentFlags().StringVar(&flags.server, "server", flags.server, "kmsvc gRPC server address (env KMSVC_SERVER)")
|
||||
root.PersistentFlags().StringVar(&flags.token, "token", "", "bearer token (env KMSVC_TOKEN)")
|
||||
root.PersistentFlags().StringVar(&flags.output, "output", flags.output, "output format: table|json (env KMSVC_OUTPUT)")
|
||||
root.PersistentFlags().BoolVar(&flags.insecure, "insecure", flags.insecure, "use plaintext gRPC instead of TLS (env KMSVC_INSECURE) — for cluster-internal/dev targets only")
|
||||
root.PersistentFlags().StringVar(&flags.clientID, "client-id", flags.clientID, "OAuth2 client_credentials client ID, used to fetch a token when --token is unset (env KMSVC_CLIENT_ID)")
|
||||
root.PersistentFlags().StringVar(&flags.clientSecret, "client-secret", "", "OAuth2 client_credentials client secret (env KMSVC_CLIENT_SECRET)")
|
||||
root.PersistentFlags().StringVar(&flags.tokenURL, "token-url", flags.tokenURL, "OAuth2 token endpoint used with --client-id/--client-secret (env KMSVC_TOKEN_URL)")
|
||||
|
||||
root.AddCommand(
|
||||
newMessageSendCmd(flags),
|
||||
newMessageReceiveCmd(flags),
|
||||
newMessageDeleteCmd(flags),
|
||||
newMessageChangeVisibilityCmd(flags),
|
||||
newDLQCmd(flags),
|
||||
newQueueCmd(flags),
|
||||
newQueueCreateCmd(flags),
|
||||
newQueueDeleteCmd(flags),
|
||||
newVersionCmd(),
|
||||
)
|
||||
|
||||
return root
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
kafkamgmtv1 "forgejo.riotpiao.homelab.com/homelab/kmsvc-proto/gen/kafkamgmt/v1"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
// fakeQueueService is a minimal QueueServiceServer for exercising CLI
|
||||
// commands end-to-end over a real (loopback) gRPC connection.
|
||||
type fakeQueueService struct {
|
||||
kafkamgmtv1.UnimplementedQueueServiceServer
|
||||
|
||||
sendMessage func(context.Context, *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error)
|
||||
receiveMessage func(context.Context, *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error)
|
||||
deleteMessage func(context.Context, *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error)
|
||||
changeMessageVisibility func(context.Context, *kafkamgmtv1.ChangeMessageVisibilityRequest) (*kafkamgmtv1.ChangeMessageVisibilityResponse, error)
|
||||
}
|
||||
|
||||
func (f *fakeQueueService) SendMessage(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
|
||||
if f.sendMessage != nil {
|
||||
return f.sendMessage(ctx, req)
|
||||
}
|
||||
return f.UnimplementedQueueServiceServer.SendMessage(ctx, req)
|
||||
}
|
||||
|
||||
func (f *fakeQueueService) ReceiveMessage(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) {
|
||||
if f.receiveMessage != nil {
|
||||
return f.receiveMessage(ctx, req)
|
||||
}
|
||||
return f.UnimplementedQueueServiceServer.ReceiveMessage(ctx, req)
|
||||
}
|
||||
|
||||
func (f *fakeQueueService) DeleteMessage(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
|
||||
if f.deleteMessage != nil {
|
||||
return f.deleteMessage(ctx, req)
|
||||
}
|
||||
return f.UnimplementedQueueServiceServer.DeleteMessage(ctx, req)
|
||||
}
|
||||
|
||||
func (f *fakeQueueService) ChangeMessageVisibility(ctx context.Context, req *kafkamgmtv1.ChangeMessageVisibilityRequest) (*kafkamgmtv1.ChangeMessageVisibilityResponse, error) {
|
||||
if f.changeMessageVisibility != nil {
|
||||
return f.changeMessageVisibility(ctx, req)
|
||||
}
|
||||
return f.UnimplementedQueueServiceServer.ChangeMessageVisibility(ctx, req)
|
||||
}
|
||||
|
||||
// startTestServer starts fake on a loopback TCP listener and returns its
|
||||
// address, registering cleanup with t.
|
||||
func startTestServer(t *testing.T, fake *fakeQueueService) string {
|
||||
t.Helper()
|
||||
|
||||
lis, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
|
||||
srv := grpc.NewServer()
|
||||
kafkamgmtv1.RegisterQueueServiceServer(srv, fake)
|
||||
go func() {
|
||||
_ = srv.Serve(lis)
|
||||
}()
|
||||
t.Cleanup(srv.Stop)
|
||||
|
||||
return lis.Addr().String()
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
// version is injected at release build time via:
|
||||
// -ldflags "-X forgejo.riotpiao.homelab.com/homelab/kmsvc-cli/internal/cli.version=v1.2.3"
|
||||
var version = "dev"
|
||||
|
||||
func newVersionCmd() *cobra.Command {
|
||||
return &cobra.Command{
|
||||
Use: "version",
|
||||
Short: "Print the kmsvc CLI version",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
fmt.Fprintln(cmd.OutOrStdout(), version)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user