Skip to content

[CF-4208] Add --filter to on-prem Flink application list #3428

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Paras Negi (paras-negi-flink) wants to merge 1 commit into from
Open

[CF-4208] Add --filter to on-prem Flink application list #3428

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion internal/flink/command_application_list.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import (
"fmt"
"strings"

"github.com/spf13/cobra"

Expand All @@ -18,6 +19,7 @@
}

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)
Expand All @@ -27,12 +29,17 @@
return cmd
}

func (c *command) applicationList(cmd *cobra.Command, _ []string) error {

Check failure on line 32 in internal/flink/command_application_list.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Refactor this method to reduce its Cognitive Complexity from 18 to the 15 allowed. 

[S3776] Cognitive Complexity of functions should not be too high
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3428&issues=617ac681-76dd-4d26-93e2-0dc720602a12&open=617ac681-76dd-4d26-93e2-0dc720602a12
environment, err := cmd.Flags().GetString("environment")
if err != nil {
return err
}

filter, err := cmd.Flags().GetString("filter")
if err != nil {
return err
}

pageSize, err := getPageSize(cmd)
if err != nil {
return err
Expand All @@ -43,7 +50,7 @@
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
}
Expand Down Expand Up @@ -82,3 +89,24 @@

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, ",")
}
29 changes: 29 additions & 0 deletions internal/flink/command_application_list_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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))
})
}
}
9 changes: 7 additions & 2 deletions pkg/flink/cmf_rest_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
1 change: 1 addition & 0 deletions test/fixtures/output/flink/application/list-env-missing.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
1 change: 1 addition & 0 deletions test/fixtures/output/flink/application/list-help-onprem.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
83 changes: 83 additions & 0 deletions test/fixtures/output/flink/application/list-name-filter-json.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
}
}
}
]
83 changes: 83 additions & 0 deletions test/fixtures/output/flink/application/list-name-status-json.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
}
}
}
]
83 changes: 83 additions & 0 deletions test/fixtures/output/flink/application/list-name-wildcard-json.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
}
}
}
]
1 change: 1 addition & 0 deletions test/fixtures/output/flink/application/list-page-size-invalid.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading