From c83835ebb97bd1450c1b07f180a7b05ae7dd82c7 Mon Sep 17 00:00:00 2001 From: Paras Negi Date: Tue, 11 Aug 2026 19:01:49 +0530 Subject: [PATCH] [CF-4208] Add `--filter` to on-prem Flink application list Add a single `--filter` flag to `confluent flink application list` that takes a CMF filter expression, for example `name=my-app*,state=RUNNING` (comma- separated `key=value` terms; `name` supports a trailing `*` wildcard). This mirrors the generic filter shape the CLI code generator will produce for CMF resources instead of inventing per-field flags. `state=` values are upper-cased before the request so users need not match CMF's case-sensitive job-state enum (`state=running` behaves like `state=RUNNING`); every other term, including case-sensitive Kubernetes `name` values, is passed through verbatim. Co-Authored-By: Claude Opus 4.8 --- internal/flink/command_application_list.go | 30 ++- .../flink/command_application_list_test.go | 29 +++ pkg/flink/cmf_rest_client.go | 9 +- .../flink/application/list-env-missing.golden | 1 + .../flink/application/list-help-onprem.golden | 1 + .../application/list-name-filter-json.golden | 83 ++++++ .../application/list-name-status-json.golden | 83 ++++++ .../list-name-wildcard-json.golden | 83 ++++++ .../application/list-page-size-invalid.golden | 1 + .../list-status-filter-json.golden | 245 ++++++++++++++++++ .../list-status-no-match-json.golden | 1 + test/flink_onprem_test.go | 9 + test/test-server/flink_onprem_handler.go | 50 ++++ 13 files changed, 622 insertions(+), 3 deletions(-) create mode 100644 internal/flink/command_application_list_test.go create mode 100644 test/fixtures/output/flink/application/list-name-filter-json.golden create mode 100644 test/fixtures/output/flink/application/list-name-status-json.golden create mode 100644 test/fixtures/output/flink/application/list-name-wildcard-json.golden create mode 100644 test/fixtures/output/flink/application/list-status-filter-json.golden create mode 100644 test/fixtures/output/flink/application/list-status-no-match-json.golden diff --git a/internal/flink/command_application_list.go b/internal/flink/command_application_list.go index 63000b8001..c837477b74 100644 --- a/internal/flink/command_application_list.go +++ b/internal/flink/command_application_list.go @@ -2,6 +2,7 @@ package flink import ( "fmt" + "strings" "github.com/spf13/cobra" @@ -18,6 +19,7 @@ func (c *command) newApplicationListCommand() *cobra.Command { } cmd.Flags().String("environment", "", "Name of the Flink environment.") + cmd.Flags().String("filter", "", `Filter the applications with a CMF filter expression, for example "name=my-app*,state=RUNNING". Terms are comma-separated "key=value" pairs; "name" accepts a trailing "*" wildcard, and "state" values are case-insensitive.`) addPageSizeFlag(cmd) addCmfFlagSet(cmd) pcmd.AddOutputFlag(cmd) @@ -33,6 +35,11 @@ func (c *command) applicationList(cmd *cobra.Command, _ []string) error { return err } + filter, err := cmd.Flags().GetString("filter") + if err != nil { + return err + } + pageSize, err := getPageSize(cmd) if err != nil { return err @@ -43,7 +50,7 @@ func (c *command) applicationList(cmd *cobra.Command, _ []string) error { return err } - applications, err := client.ListApplications(c.createContext(), environment, pageSize) + applications, err := client.ListApplications(c.createContext(), environment, normalizeApplicationFilter(filter), pageSize) if err != nil { return err } @@ -82,3 +89,24 @@ func (c *command) applicationList(cmd *cobra.Command, _ []string) error { return output.SerializedOutput(cmd, localApps) } + +// normalizeApplicationFilter case-folds the value of any "state" term in a CMF applications +// filter expression to upper case. Flink job states are an upper-case enum and CMF matches +// them case-sensitively, so this lets users write "state=running" instead of "state=RUNNING". +// Every other term is passed through verbatim: "name" values in particular are case-sensitive +// (Kubernetes resource names), so they must not be folded. The grammar is the comma-separated +// "key=value" syntax of the CMF applications list "filter" query parameter. +func normalizeApplicationFilter(filter string) string { + if filter == "" { + return "" + } + + terms := strings.Split(filter, ",") + for i, term := range terms { + key, value, found := strings.Cut(term, "=") + if found && strings.EqualFold(strings.TrimSpace(key), "state") { + terms[i] = "state=" + strings.ToUpper(strings.TrimSpace(value)) + } + } + return strings.Join(terms, ",") +} diff --git a/internal/flink/command_application_list_test.go b/internal/flink/command_application_list_test.go new file mode 100644 index 0000000000..cea59db0dc --- /dev/null +++ b/internal/flink/command_application_list_test.go @@ -0,0 +1,29 @@ +package flink + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNormalizeApplicationFilter(t *testing.T) { + tests := []struct { + name string + filter string + want string + }{ + {name: "empty", filter: "", want: ""}, + {name: "name is not folded", filter: "name=My-App*", want: "name=My-App*"}, + {name: "state lower-cased is folded", filter: "state=running", want: "state=RUNNING"}, + {name: "state already upper", filter: "state=RUNNING", want: "state=RUNNING"}, + {name: "state key case-insensitive", filter: "State=running", want: "state=RUNNING"}, + {name: "name and state", filter: "name=My-App*,state=reconciling", want: "name=My-App*,state=RECONCILING"}, + {name: "unknown key passed through", filter: "foo=Bar", want: "foo=Bar"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.want, normalizeApplicationFilter(test.filter)) + }) + } +} diff --git a/pkg/flink/cmf_rest_client.go b/pkg/flink/cmf_rest_client.go index 83a76a0672..b5856f1b27 100644 --- a/pkg/flink/cmf_rest_client.go +++ b/pkg/flink/cmf_rest_client.go @@ -189,9 +189,14 @@ func (cmfClient *CmfRestClient) DescribeApplication(ctx context.Context, environ return cmfApplication, nil } -func (cmfClient *CmfRestClient) ListApplications(ctx context.Context, environment string, pageSize int32) ([]cmfsdk.FlinkApplication, error) { +func (cmfClient *CmfRestClient) ListApplications(ctx context.Context, environment, filter string, pageSize int32) ([]cmfsdk.FlinkApplication, error) { + request := cmfClient.FlinkApplicationsApi.GetApplications(ctx, environment) + if filter != "" { + request = request.Filter(filter) + } + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.FlinkApplication, error) { - applicationsPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplications(ctx, environment).Page(page).Size(size).Execute() + applicationsPage, httpResponse, err := request.Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list applications in the environment "%s": %s`, environment, parsedErr) } diff --git a/test/fixtures/output/flink/application/list-env-missing.golden b/test/fixtures/output/flink/application/list-env-missing.golden index 08ba04a4e9..7249dd6697 100644 --- a/test/fixtures/output/flink/application/list-env-missing.golden +++ b/test/fixtures/output/flink/application/list-env-missing.golden @@ -4,6 +4,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --filter string Filter the applications with a CMF filter expression, for example "name=my-app*,state=RUNNING". Terms are comma-separated "key=value" pairs; "name" accepts a trailing "*" wildcard, and "state" values are case-insensitive. --page-size int32 Number of results to fetch per API request while paginating; does not cap the total results returned. (default 100) --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-help-onprem.golden b/test/fixtures/output/flink/application/list-help-onprem.golden index 9cfa6a2791..72aea8e6b1 100644 --- a/test/fixtures/output/flink/application/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/list-help-onprem.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --filter string Filter the applications with a CMF filter expression, for example "name=my-app*,state=RUNNING". Terms are comma-separated "key=value" pairs; "name" accepts a trailing "*" wildcard, and "state" values are case-insensitive. --page-size int32 Number of results to fetch per API request while paginating; does not cap the total results returned. (default 100) --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-name-filter-json.golden b/test/fixtures/output/flink/application/list-name-filter-json.golden new file mode 100644 index 0000000000..273d16e858 --- /dev/null +++ b/test/fixtures/output/flink/application/list-name-filter-json.golden @@ -0,0 +1,83 @@ +[ + { + "apiVersion": "cmf.confluent.io/v1", + "kind": "FlinkApplication", + "metadata": { + "name": "default-application-s" + }, + "spec": { + "flinkConfiguration": { + "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory", + "metrics.reporter.prom.port": "9249-9250", + "taskmanager.numberOfTaskSlots": "8" + }, + "flinkVersion": "v1_19", + "image": "confluentinc/cp-flink:1.19.1-cp1", + "job": { + "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar", + "parallelism": 3, + "state": "running", + "upgradeMode": "stateless" + }, + "jobManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + }, + "serviceAccount": "flink", + "taskManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + } + }, + "status": { + "clusterInfo": { + "flink-revision": "89d0b8f @ 2024-06-22T13:19:31+02:00", + "flink-version": "1.19.1-cp1", + "total-cpu": "3.0", + "total-memory": "3296722944" + }, + "error": null, + "jobManagerDeploymentStatus": "DEPLOYING", + "jobStatus": { + "checkpointInfo": { + "formatType": null, + "lastCheckpoint": null, + "lastPeriodicCheckpointTimestamp": 0, + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "jobId": "dcabb1ad6c40495bc2d7fa7a0097c5aa", + "jobName": "State machine job", + "savepointInfo": { + "formatType": null, + "lastPeriodicSavepointTimestamp": 0, + "lastSavepoint": null, + "savepointHistory": [], + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "startTime": "1726640263746", + "state": "RECONCILING", + "updateTime": "1726640280561" + }, + "lifecycleState": "DEPLOYED", + "observedGeneration": 4, + "reconciliationStatus": { + "lastReconciledSpec": "", + "lastStableSpec": "", + "reconciliationTimestamp": 1726640346899, + "state": "DEPLOYED" + }, + "taskManager": { + "labelSelector": "component=taskmanager,app=basic-example", + "replicas": 1 + } + } + } +] diff --git a/test/fixtures/output/flink/application/list-name-status-json.golden b/test/fixtures/output/flink/application/list-name-status-json.golden new file mode 100644 index 0000000000..230f62ffdc --- /dev/null +++ b/test/fixtures/output/flink/application/list-name-status-json.golden @@ -0,0 +1,83 @@ +[ + { + "apiVersion": "cmf.confluent.io/v1", + "kind": "FlinkApplication", + "metadata": { + "name": "default-application-1" + }, + "spec": { + "flinkConfiguration": { + "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory", + "metrics.reporter.prom.port": "9249-9250", + "taskmanager.numberOfTaskSlots": "8" + }, + "flinkVersion": "v1_19", + "image": "confluentinc/cp-flink:1.19.1-cp1", + "job": { + "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar", + "parallelism": 3, + "state": "running", + "upgradeMode": "stateless" + }, + "jobManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + }, + "serviceAccount": "flink", + "taskManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + } + }, + "status": { + "clusterInfo": { + "flink-revision": "89d0b8f @ 2024-06-22T13:19:31+02:00", + "flink-version": "1.19.1-cp1", + "total-cpu": "3.0", + "total-memory": "3296722944" + }, + "error": null, + "jobManagerDeploymentStatus": "DEPLOYING", + "jobStatus": { + "checkpointInfo": { + "formatType": null, + "lastCheckpoint": null, + "lastPeriodicCheckpointTimestamp": 0, + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "jobId": "dcabb1ad6c40495bc2d7fa7a0097c5aa", + "jobName": "State machine job", + "savepointInfo": { + "formatType": null, + "lastPeriodicSavepointTimestamp": 0, + "lastSavepoint": null, + "savepointHistory": [], + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "startTime": "1726640263746", + "state": "RECONCILING", + "updateTime": "1726640280561" + }, + "lifecycleState": "DEPLOYED", + "observedGeneration": 4, + "reconciliationStatus": { + "lastReconciledSpec": "", + "lastStableSpec": "", + "reconciliationTimestamp": 1726640346899, + "state": "DEPLOYED" + }, + "taskManager": { + "labelSelector": "component=taskmanager,app=basic-example", + "replicas": 1 + } + } + } +] diff --git a/test/fixtures/output/flink/application/list-name-wildcard-json.golden b/test/fixtures/output/flink/application/list-name-wildcard-json.golden new file mode 100644 index 0000000000..230f62ffdc --- /dev/null +++ b/test/fixtures/output/flink/application/list-name-wildcard-json.golden @@ -0,0 +1,83 @@ +[ + { + "apiVersion": "cmf.confluent.io/v1", + "kind": "FlinkApplication", + "metadata": { + "name": "default-application-1" + }, + "spec": { + "flinkConfiguration": { + "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory", + "metrics.reporter.prom.port": "9249-9250", + "taskmanager.numberOfTaskSlots": "8" + }, + "flinkVersion": "v1_19", + "image": "confluentinc/cp-flink:1.19.1-cp1", + "job": { + "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar", + "parallelism": 3, + "state": "running", + "upgradeMode": "stateless" + }, + "jobManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + }, + "serviceAccount": "flink", + "taskManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + } + }, + "status": { + "clusterInfo": { + "flink-revision": "89d0b8f @ 2024-06-22T13:19:31+02:00", + "flink-version": "1.19.1-cp1", + "total-cpu": "3.0", + "total-memory": "3296722944" + }, + "error": null, + "jobManagerDeploymentStatus": "DEPLOYING", + "jobStatus": { + "checkpointInfo": { + "formatType": null, + "lastCheckpoint": null, + "lastPeriodicCheckpointTimestamp": 0, + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "jobId": "dcabb1ad6c40495bc2d7fa7a0097c5aa", + "jobName": "State machine job", + "savepointInfo": { + "formatType": null, + "lastPeriodicSavepointTimestamp": 0, + "lastSavepoint": null, + "savepointHistory": [], + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "startTime": "1726640263746", + "state": "RECONCILING", + "updateTime": "1726640280561" + }, + "lifecycleState": "DEPLOYED", + "observedGeneration": 4, + "reconciliationStatus": { + "lastReconciledSpec": "", + "lastStableSpec": "", + "reconciliationTimestamp": 1726640346899, + "state": "DEPLOYED" + }, + "taskManager": { + "labelSelector": "component=taskmanager,app=basic-example", + "replicas": 1 + } + } + } +] diff --git a/test/fixtures/output/flink/application/list-page-size-invalid.golden b/test/fixtures/output/flink/application/list-page-size-invalid.golden index 064ba001f7..eb2655d2a9 100644 --- a/test/fixtures/output/flink/application/list-page-size-invalid.golden +++ b/test/fixtures/output/flink/application/list-page-size-invalid.golden @@ -4,6 +4,7 @@ Usage: Flags: --environment string Name of the Flink environment. + --filter string Filter the applications with a CMF filter expression, for example "name=my-app*,state=RUNNING". Terms are comma-separated "key=value" pairs; "name" accepts a trailing "*" wildcard, and "state" values are case-insensitive. --page-size int32 Number of results to fetch per API request while paginating; does not cap the total results returned. (default 100) --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-status-filter-json.golden b/test/fixtures/output/flink/application/list-status-filter-json.golden new file mode 100644 index 0000000000..fee5108863 --- /dev/null +++ b/test/fixtures/output/flink/application/list-status-filter-json.golden @@ -0,0 +1,245 @@ +[ + { + "apiVersion": "cmf.confluent.io/v1", + "kind": "FlinkApplication", + "metadata": { + "name": "default-application-1" + }, + "spec": { + "flinkConfiguration": { + "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory", + "metrics.reporter.prom.port": "9249-9250", + "taskmanager.numberOfTaskSlots": "8" + }, + "flinkVersion": "v1_19", + "image": "confluentinc/cp-flink:1.19.1-cp1", + "job": { + "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar", + "parallelism": 3, + "state": "running", + "upgradeMode": "stateless" + }, + "jobManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + }, + "serviceAccount": "flink", + "taskManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + } + }, + "status": { + "clusterInfo": { + "flink-revision": "89d0b8f @ 2024-06-22T13:19:31+02:00", + "flink-version": "1.19.1-cp1", + "total-cpu": "3.0", + "total-memory": "3296722944" + }, + "error": null, + "jobManagerDeploymentStatus": "DEPLOYING", + "jobStatus": { + "checkpointInfo": { + "formatType": null, + "lastCheckpoint": null, + "lastPeriodicCheckpointTimestamp": 0, + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "jobId": "dcabb1ad6c40495bc2d7fa7a0097c5aa", + "jobName": "State machine job", + "savepointInfo": { + "formatType": null, + "lastPeriodicSavepointTimestamp": 0, + "lastSavepoint": null, + "savepointHistory": [], + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "startTime": "1726640263746", + "state": "RECONCILING", + "updateTime": "1726640280561" + }, + "lifecycleState": "DEPLOYED", + "observedGeneration": 4, + "reconciliationStatus": { + "lastReconciledSpec": "", + "lastStableSpec": "", + "reconciliationTimestamp": 1726640346899, + "state": "DEPLOYED" + }, + "taskManager": { + "labelSelector": "component=taskmanager,app=basic-example", + "replicas": 1 + } + } + }, + { + "apiVersion": "cmf.confluent.io/v1", + "kind": "FlinkApplication", + "metadata": { + "name": "default-application-2" + }, + "spec": { + "flinkConfiguration": { + "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory", + "metrics.reporter.prom.port": "9249-9250", + "taskmanager.numberOfTaskSlots": "8" + }, + "flinkVersion": "v1_19", + "image": "confluentinc/cp-flink:1.19.1-cp1", + "job": { + "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar", + "parallelism": 3, + "state": "running", + "upgradeMode": "stateless" + }, + "jobManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + }, + "serviceAccount": "flink", + "taskManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + } + }, + "status": { + "clusterInfo": { + "flink-revision": "89d0b8f @ 2024-06-22T13:19:31+02:00", + "flink-version": "1.19.1-cp1", + "total-cpu": "3.0", + "total-memory": "3296722944" + }, + "error": null, + "jobManagerDeploymentStatus": "DEPLOYING", + "jobStatus": { + "checkpointInfo": { + "formatType": null, + "lastCheckpoint": null, + "lastPeriodicCheckpointTimestamp": 0, + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "jobId": "dcabb1ad6c40495bc2d7fa7a0097c5aa", + "jobName": "State machine job", + "savepointInfo": { + "formatType": null, + "lastPeriodicSavepointTimestamp": 0, + "lastSavepoint": null, + "savepointHistory": [], + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "startTime": "1726640263746", + "state": "RECONCILING", + "updateTime": "1726640280561" + }, + "lifecycleState": "DEPLOYED", + "observedGeneration": 4, + "reconciliationStatus": { + "lastReconciledSpec": "", + "lastStableSpec": "", + "reconciliationTimestamp": 1726640346899, + "state": "DEPLOYED" + }, + "taskManager": { + "labelSelector": "component=taskmanager,app=basic-example", + "replicas": 1 + } + } + }, + { + "apiVersion": "cmf.confluent.io/v1", + "kind": "FlinkApplication", + "metadata": { + "name": "default-application-s" + }, + "spec": { + "flinkConfiguration": { + "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory", + "metrics.reporter.prom.port": "9249-9250", + "taskmanager.numberOfTaskSlots": "8" + }, + "flinkVersion": "v1_19", + "image": "confluentinc/cp-flink:1.19.1-cp1", + "job": { + "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar", + "parallelism": 3, + "state": "running", + "upgradeMode": "stateless" + }, + "jobManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + }, + "serviceAccount": "flink", + "taskManager": { + "resource": { + "cpu": 1, + "memory": "1048m" + } + } + }, + "status": { + "clusterInfo": { + "flink-revision": "89d0b8f @ 2024-06-22T13:19:31+02:00", + "flink-version": "1.19.1-cp1", + "total-cpu": "3.0", + "total-memory": "3296722944" + }, + "error": null, + "jobManagerDeploymentStatus": "DEPLOYING", + "jobStatus": { + "checkpointInfo": { + "formatType": null, + "lastCheckpoint": null, + "lastPeriodicCheckpointTimestamp": 0, + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "jobId": "dcabb1ad6c40495bc2d7fa7a0097c5aa", + "jobName": "State machine job", + "savepointInfo": { + "formatType": null, + "lastPeriodicSavepointTimestamp": 0, + "lastSavepoint": null, + "savepointHistory": [], + "triggerId": null, + "triggerTimestamp": null, + "triggerType": null + }, + "startTime": "1726640263746", + "state": "RECONCILING", + "updateTime": "1726640280561" + }, + "lifecycleState": "DEPLOYED", + "observedGeneration": 4, + "reconciliationStatus": { + "lastReconciledSpec": "", + "lastStableSpec": "", + "reconciliationTimestamp": 1726640346899, + "state": "DEPLOYED" + }, + "taskManager": { + "labelSelector": "component=taskmanager,app=basic-example", + "replicas": 1 + } + } + } +] diff --git a/test/fixtures/output/flink/application/list-status-no-match-json.golden b/test/fixtures/output/flink/application/list-status-no-match-json.golden new file mode 100644 index 0000000000..fe51488c70 --- /dev/null +++ b/test/fixtures/output/flink/application/list-status-no-match-json.golden @@ -0,0 +1 @@ +[] diff --git a/test/flink_onprem_test.go b/test/flink_onprem_test.go index 15a0ab33cc..7b3cb9aebd 100644 --- a/test/flink_onprem_test.go +++ b/test/flink_onprem_test.go @@ -40,6 +40,15 @@ func (s *CLITestSuite) TestFlinkApplicationList() { {args: "flink application list --environment default --page-size 2 --output json", fixture: "flink/application/list-json.golden"}, // a non-positive page size falls back to the default and still returns the full list {args: "flink application list --environment default --page-size -1 --output json", fixture: "flink/application/list-json.golden"}, + // filtering: --filter passes a CMF filter expression through; "state" values are case-folded + {args: "flink application list --environment default --filter name=default-application-s --output json", fixture: "flink/application/list-name-filter-json.golden"}, + {args: "flink application list --environment default --filter name=default-application-1* --output json", fixture: "flink/application/list-name-wildcard-json.golden"}, + // lower-case and upper-case "state" resolve to the same result, proving the case-folding + {args: "flink application list --environment default --filter state=reconciling --output json", fixture: "flink/application/list-status-filter-json.golden"}, + {args: "flink application list --environment default --filter state=RECONCILING --output json", fixture: "flink/application/list-status-filter-json.golden"}, + // an unknown state is forwarded as-is and simply matches nothing (no client-side validation) + {args: "flink application list --environment default --filter state=bogus --output json", fixture: "flink/application/list-status-no-match-json.golden"}, + {args: "flink application list --environment default --filter name=default-application-1*,state=reconciling --output json", fixture: "flink/application/list-name-status-json.golden"}, } runIntegrationTestsWithMultipleAuth(s, tests) diff --git a/test/test-server/flink_onprem_handler.go b/test/test-server/flink_onprem_handler.go index 655a600d78..2d8753800d 100644 --- a/test/test-server/flink_onprem_handler.go +++ b/test/test-server/flink_onprem_handler.go @@ -146,6 +146,55 @@ func paginateApplications(all []cmfsdk.FlinkApplication, pageParam, sizeParam st return all[start:end] } +// filterApplications emulates the CMF server-side "filter" query param for the applications +// endpoint. It understands comma-separated "name=" (with optional "*" suffix wildcard) +// and "state=" expressions, as passed by the CLI's --filter flag. +func filterApplications(items []cmfsdk.FlinkApplication, filter string) []cmfsdk.FlinkApplication { + if filter == "" { + return items + } + + for _, expr := range strings.Split(filter, ",") { + key, value, found := strings.Cut(expr, "=") + if !found { + continue + } + matched := make([]cmfsdk.FlinkApplication, 0, len(items)) + for _, item := range items { + if applicationMatchesFilter(item, key, value) { + matched = append(matched, item) + } + } + items = matched + } + return items +} + +func applicationMatchesFilter(app cmfsdk.FlinkApplication, key, value string) bool { + switch key { + case "name": + name, _ := app.Metadata["name"].(string) + if prefix, isWildcard := strings.CutSuffix(value, "*"); isWildcard { + return strings.HasPrefix(name, prefix) + } + return name == value + case "state": + if app.Status == nil { + return false + } + jobStatus, ok := (*app.Status)["jobStatus"].(map[string]interface{}) + if !ok { + return false + } + // CMF matches state case-sensitively against the upper-case job-state enum; the CLI + // upper-cases the user's "state=" value before sending, so the match here is exact. + state, _ := jobStatus["state"].(string) + return state == value + default: + return true + } +} + // Helper function to create a Flink environment. func createEnvironment(name string, namespace string) cmfsdk.Environment { createdTime := time.Date(2024, time.September, 10, 23, 0, 0, 0, time.UTC) @@ -725,6 +774,7 @@ func handleCmfApplications(t *testing.T) http.HandlerFunc { allItems = []cmfsdk.FlinkApplication{createApplication("update-failure-application")} } + allItems = filterApplications(allItems, r.URL.Query().Get("filter")) applicationsPage := map[string]interface{}{ "items": paginateApplications(allItems, r.URL.Query().Get("page"), r.URL.Query().Get("size")), }