Add Web UI information to status

pkg/apis/app/v1beta1/types.go

@@ -127,6 +127,7 @@ type SavepointInfo struct {
 }
 
 type FlinkClusterStatus struct {
+	ClusterOverviewURL   string       `json:"clusterOverviewURL,omitempty"`
 	Health               HealthStatus `json:"health,omitempty"`
 	NumberOfTaskManagers int32        `json:"numberOfTaskManagers,omitempty"`
 	HealthyTaskManagers  int32        `json:"healthyTaskManagers,omitepty"`
@@ -135,9 +136,10 @@ type FlinkClusterStatus struct {
 }
 
 type FlinkJobStatus struct {
-	JobID  string       `json:"jobID,omitEmpty"`
-	Health HealthStatus `json:"health,omitEmpty"`
-	State  JobState     `json:"state,omitEmpty"`
+	JobOverviewURL string       `json:"jobOverviewURL,omitempty"`
+	JobID          string       `json:"jobID,omitEmpty"`
+	Health         HealthStatus `json:"health,omitEmpty"`
+	State          JobState     `json:"state,omitEmpty"`
 
 	JarName               string `json:"jarName"`
 	Parallelism           int32  `json:"parallelism"`

pkg/controller/flink/client/api.go

@@ -16,14 +16,16 @@ import (
 	"github.com/pkg/errors"
 )
 
+const GetJobsOverviewURL = "/jobs/%s"
+const GetClusterOverviewURL = "/overview"
+const WebUIAnchor = "/#"
+
 const submitJobURL = "/jars/%s/run"
 const savepointURL = "/jobs/%s/savepoints"
 const jobURL = "/jobs/%s"
 const checkSavepointStatusURL = "/jobs/%s/savepoints/%s"
 const getJobsURL = "/jobs"
-const getJobsOverviewURL = "/jobs/%s"
 const getJobConfigURL = "/jobs/%s/config"
-const getOverviewURL = "/overview"
 const checkpointsURL = "/jobs/%s/checkpoints"
 const taskmanagersURL = "/taskmanagers"
 const httpGet = "GET"
@@ -119,7 +121,7 @@ func (c *FlinkJobManagerClient) GetJobConfig(ctx context.Context, url, jobID str
 }
 
 func (c *FlinkJobManagerClient) GetClusterOverview(ctx context.Context, url string) (*ClusterOverviewResponse, error) {
-	url = url + getOverviewURL
+	url = url + GetClusterOverviewURL
 	response, err := c.executeRequest(ctx, httpGet, url, nil)
 	if err != nil {
 		c.metrics.getClusterFailureCounter.Inc(ctx)
@@ -347,7 +349,7 @@ func (c *FlinkJobManagerClient) GetCheckpointCounts(ctx context.Context, url str
 }
 
 func (c *FlinkJobManagerClient) GetJobOverview(ctx context.Context, url string, jobID string) (*FlinkJobOverview, error) {
-	endpoint := fmt.Sprintf(url+getJobsOverviewURL, jobID)
+	endpoint := fmt.Sprintf(url+GetJobsOverviewURL, jobID)
 	response, err := c.executeRequest(ctx, httpGet, endpoint, nil)
 	if err != nil {
 		return nil, GetRetryableError(err, GetJobOverview, GlobalFailure, DefaultRetries)

pkg/controller/flink/flink.go

@@ -140,6 +140,15 @@ func getURLFromApp(application *v1beta1.FlinkApplication, hash string) string {
 	return fmt.Sprintf("http://%s.%s:%d", service, application.Namespace, port)
 }
 
+func getExternalURLFromApp(application *v1beta1.FlinkApplication) string {
+	cfg := controllerConfig.GetConfig()
+	// Local environment
+	if cfg.UseProxy {
+		return fmt.Sprintf(proxyURL, cfg.ProxyPort.Port, application.Namespace, application.Name)
+	}
+	return GetFlinkUIIngressURL(application.Name)
+}
+
 func GetActiveFlinkJob(jobs []client.FlinkJob) *client.FlinkJob {
 	if len(jobs) == 0 {
 		return nil
@@ -433,7 +442,7 @@ func (f *Controller) CompareAndUpdateClusterStatus(ctx context.Context, applicat
 	if deployment == nil || err != nil {
 		return false, err
 	}
-
+	application.Status.ClusterStatus.ClusterOverviewURL = fmt.Sprintf(getExternalURLFromApp(application) + client.WebUIAnchor + client.GetClusterOverviewURL)
 	application.Status.ClusterStatus.NumberOfTaskManagers = deployment.Taskmanager.Status.AvailableReplicas
 	// Get Cluster overview
 	response, err := f.flinkClient.GetClusterOverview(ctx, getURLFromApp(application, hash))
@@ -484,7 +493,6 @@ func (f *Controller) CompareAndUpdateJobStatus(ctx context.Context, app *v1beta1
 	}
 
 	oldJobStatus := app.Status.JobStatus
-
 	app.Status.JobStatus.JobID = oldJobStatus.JobID
 	jobResponse, err := f.flinkClient.GetJobOverview(ctx, getURLFromApp(app, hash), app.Status.JobStatus.JobID)
 	if err != nil {
@@ -496,6 +504,7 @@ func (f *Controller) CompareAndUpdateJobStatus(ctx context.Context, app *v1beta1
 	}
 
 	// Job status
+	app.Status.JobStatus.JobOverviewURL = fmt.Sprintf(getExternalURLFromApp(app)+client.WebUIAnchor+client.GetJobsOverviewURL, app.Status.JobStatus.JobID)
 	app.Status.JobStatus.State = v1beta1.JobState(jobResponse.State)
 	jobStartTime := metav1.NewTime(time.Unix(jobResponse.StartTime/1000, 0))
 	app.Status.JobStatus.StartTime = &jobStartTime

pkg/controller/flink/flink_test.go

@@ -635,6 +635,7 @@ func TestClusterStatusUpdated(t *testing.T) {
 	assert.Equal(t, int32(0), flinkApp.Status.ClusterStatus.AvailableTaskSlots)
 	assert.Equal(t, int32(1), flinkApp.Status.ClusterStatus.HealthyTaskManagers)
 	assert.Equal(t, v1beta1.Green, flinkApp.Status.ClusterStatus.Health)
+	assert.Equal(t, "/#/overview", flinkApp.Status.ClusterStatus.ClusterOverviewURL)
 
 }
 
@@ -791,6 +792,7 @@ func TestJobStatusUpdated(t *testing.T) {
 	assert.Equal(t, &expectedTime, flinkApp.Status.JobStatus.RestoreTime)
 	assert.Equal(t, "/test/externalpath", flinkApp.Status.JobStatus.RestorePath)
 	assert.Equal(t, &expectedTime, flinkApp.Status.JobStatus.LastCheckpointTime)
+	assert.Equal(t, "/#/jobs/abc", flinkApp.Status.JobStatus.JobOverviewURL)
 
 }
 
@@ -810,6 +812,7 @@ func TestNoJobStatusChange(t *testing.T) {
 	app1.Status.JobStatus.Health = v1beta1.Green
 	app1.Status.JobStatus.RestoreTime = &metaTime
 	app1.Status.JobStatus.RestorePath = "/test/externalpath"
+	app1.Status.JobStatus.JobOverviewURL = "/#/jobs/j1"
 
 	mockJmClient.GetJobOverviewFunc = func(ctx context.Context, url string, jobID string) (*client.FlinkJobOverview, error) {
 		assert.Equal(t, url, "http://app-name-hash.ns:8081")