func (observer *ClusterStateObserver) observeJobPod(
	observedPod *corev1.Pod) error {
	var log = observer.log
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	var podSelector = labels.SelectorFromSet(map[string]string{"job-name": getJobName(clusterName)})
	var podList = new(corev1.PodList)
	var err = observer.k8sClient.List(
		observer.context,
		podList,
		client.InNamespace(clusterNamespace),
		client.MatchingLabelsSelector{Selector: podSelector})
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get job submitter pod list")
			return err
		}
		log.Info("Observed job submitter pod list", "state", "nil")
	} else {
		log.Info("Observed job submitter pod list", "state", *podList)
	}
	if podList != nil && len(podList.Items) > 0 {
		podList.Items[0].DeepCopyInto(observedPod)
	}
	return nil
}
observeJobPod observes job submitter pod.
package controllers
import (
	"context"
	"errors"
	"fmt"
	"github.com/go-logr/logr"
	v1beta1 "github.com/googlecloudplatform/flink-operator/api/v1beta1"
	"github.com/googlecloudplatform/flink-operator/controllers/flinkclient"
	"github.com/googlecloudplatform/flink-operator/controllers/history"
	appsv1 "k8s.io/api/apps/v1"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"strings"
	"time"
)
type ClusterStateObserver struct {
	k8sClient   client.Client
	flinkClient flinkclient.FlinkClient
	request     ctrl.Request
	context     context.Context
	log         logr.Logger
	history     history.Interface
}
type ObservedClusterState struct {
	cluster           *v1beta1.FlinkCluster
	revisions         []*appsv1.ControllerRevision
	configMap         *corev1.ConfigMap
	jmStatefulSet      *appsv1.StatefulSet
	jmService         *corev1.Service
	jmIngress         *extensionsv1beta1.Ingress
	tmStatefulSet      *appsv1.StatefulSet
	job               *batchv1.Job
	jobPod            *corev1.Pod
	flinkJobStatus    FlinkJobStatus
	flinkJobSubmitLog *FlinkJobSubmitLog
	savepoint         *flinkclient.SavepointStatus
	revisionStatus    *RevisionStatus
	savepointErr      error
	observeTime       time.Time
}
type FlinkJobStatus struct {
	flinkJob            *flinkclient.JobStatus
	flinkJobList        *flinkclient.JobStatusList
	flinkJobsUnexpected []string
}
type FlinkJobSubmitLog struct {
	JobID   string `yaml:"jobID,omitempty"`
	Message string `yaml:"message"`
}
func (observer *ClusterStateObserver) observe(
	observed *ObservedClusterState) error {
	var err error
	var log = observer.log
	var observedCluster = new(v1beta1.FlinkCluster)
	err = observer.observeCluster(observedCluster)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get the cluster resource")
			return err
		}
		log.Info("Observed cluster", "cluster", "nil")
		observedCluster = nil
	} else {
		log.Info("Observed cluster", "cluster", *observedCluster)
		observed.cluster = observedCluster
	}
	var observedRevisions []*appsv1.ControllerRevision
	err = observer.observeRevisions(&observedRevisions, observedCluster)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get the controllerRevision resource list")
			return err
		}
		log.Info("Observed controllerRevisions", "controllerRevisions", "nil")
	} else {
		var b strings.Builder
		for _, cr := range observedRevisions {
			fmt.Fprintf(&b, "{name: %v, revision: %v},", cr.Name, cr.Revision)
		}
		log.Info("Observed controllerRevisions", "controllerRevisions", fmt.Sprintf("[%v]", b.String()))
		observed.revisions = observedRevisions
	}
	var observedConfigMap = new(corev1.ConfigMap)
	err = observer.observeConfigMap(observedConfigMap)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get configMap")
			return err
		}
		log.Info("Observed configMap", "state", "nil")
		observedConfigMap = nil
	} else {
		log.Info("Observed configMap", "state", *observedConfigMap)
		observed.configMap = observedConfigMap
	}
	var observedJmStatefulSet = new(appsv1.StatefulSet)
	err = observer.observeJobManagerStatefulSet(observedJmStatefulSet)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get JobManager StatefulSet")
			return err
		}
		log.Info("Observed JobManager StatefulSet", "state", "nil")
		observedJmStatefulSet = nil
	} else {
		log.Info("Observed JobManager StatefulSet", "state", *observedJmStatefulSet)
		observed.jmStatefulSet = observedJmStatefulSet
	}
	var observedJmService = new(corev1.Service)
	err = observer.observeJobManagerService(observedJmService)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get JobManager service")
			return err
		}
		log.Info("Observed JobManager service", "state", "nil")
		observedJmService = nil
	} else {
		log.Info("Observed JobManager service", "state", *observedJmService)
		observed.jmService = observedJmService
	}
	var observedJmIngress = new(extensionsv1beta1.Ingress)
	err = observer.observeJobManagerIngress(observedJmIngress)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get JobManager ingress")
			return err
		}
		log.Info("Observed JobManager ingress", "state", "nil")
		observedJmIngress = nil
	} else {
		log.Info("Observed JobManager ingress", "state", *observedJmIngress)
		observed.jmIngress = observedJmIngress
	}
	var observedTmStatefulSet = new(appsv1.StatefulSet)
	err = observer.observeTaskManagerStatefulSet(observedTmStatefulSet)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get TaskManager StatefulSet")
			return err
		}
		log.Info("Observed TaskManager StatefulSet", "state", "nil")
		observedTmStatefulSet = nil
	} else {
		log.Info("Observed TaskManager StatefulSet", "state", *observedTmStatefulSet)
		observed.tmStatefulSet = observedTmStatefulSet
	}
	observer.observeSavepoint(observed)
	err = observer.observeJob(observed)
	observed.observeTime = time.Now()
	return err
}
func (observer *ClusterStateObserver) observeJob(
	observed *ObservedClusterState) error {
	if observed.cluster == nil {
		return nil
	}
	var observedJob *batchv1.Job
	var observedFlinkJobStatus FlinkJobStatus
	var observedFlinkJobSubmitLog *FlinkJobSubmitLog
	var recordedJobStatus = observed.cluster.Status.Components.Job
	var err error
	var log = observer.log
	if observed.cluster == nil || observed.cluster.Spec.Job == nil {
		return nil
	}
	observedJob = new(batchv1.Job)
	err = observer.observeJobResource(observedJob)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get job")
			return err
		}
		log.Info("Observed job submitter", "state", "nil")
		observedJob = nil
	} else {
		log.Info("Observed job submitter", "state", *observedJob)
	}
	observed.job = observedJob
	var jobSubmitCompleted = observedJob != nil && (observedJob.Status.Succeeded > 0 || observedJob.Status.Failed > 0)
	var jobInPendingState = recordedJobStatus != nil && recordedJobStatus.State == v1beta1.JobStatePending
	var flinkJobID string
	if jobSubmitCompleted && jobInPendingState {
		var observedJobPod *corev1.Pod
		observedJobPod = new(corev1.Pod)
		err = observer.observeJobPod(observedJobPod)
		if err != nil {
			log.Error(err, "Failed to get job pod")
		}
		observed.jobPod = observedJobPod
		observedFlinkJobSubmitLog, err = getFlinkJobSubmitLog(observedJobPod)
		if err != nil {
			log.Error(err, "Failed to extract job submit result")
		}
		if observedFlinkJobSubmitLog != nil && observedFlinkJobSubmitLog.JobID != "" {
			flinkJobID = observedFlinkJobSubmitLog.JobID
		}
		observed.flinkJobSubmitLog = observedFlinkJobSubmitLog
	}
	if flinkJobID == "" && recordedJobStatus != nil {
		flinkJobID = recordedJobStatus.ID
	}
	observer.observeFlinkJobStatus(observed, flinkJobID, &observedFlinkJobStatus)
	observed.flinkJobStatus = observedFlinkJobStatus
	return nil
}
func (observer *ClusterStateObserver) observeFlinkJobStatus(
	observed *ObservedClusterState,
	flinkJobID string,
	flinkJobStatus *FlinkJobStatus) {
	var log = observer.log
	var flinkJob *flinkclient.JobStatus
	var flinkJobList *flinkclient.JobStatusList
	var flinkJobsUnexpected []string
	var jmReady = observed.jmStatefulSet != nil && getStatefulSetState(observed.jmStatefulSet) == v1beta1.ComponentStateReady
	if !jmReady {
		log.Info(
			"Skip getting Flink job status.",
			"job manager ready", jmReady)
		return
	}
	flinkJobList = &flinkclient.JobStatusList{}
	var flinkAPIBaseURL = getFlinkAPIBaseURL(observed.cluster)
	var err = observer.flinkClient.GetJobStatusList(flinkAPIBaseURL, flinkJobList)
	if err != nil {
		log.Info("Failed to get Flink job status list.", "error", err)
		return
	}
	log.Info("Observed Flink job status list", "jobs", flinkJobList.Jobs)
	flinkJobStatus.flinkJobList = flinkJobList
	if flinkJobID == "" {
		return
	}
	for _, job := range flinkJobList.Jobs {
		if flinkJobID == job.ID {
			flinkJob = new(flinkclient.JobStatus)
			*flinkJob = job
		} else if getFlinkJobDeploymentState(job.Status) == v1beta1.JobStateRunning {
			flinkJobsUnexpected = append(flinkJobsUnexpected, job.ID)
		}
	}
	flinkJobStatus.flinkJob = flinkJob
	flinkJobStatus.flinkJobsUnexpected = flinkJobsUnexpected
	if len(flinkJobsUnexpected) > 1 {
		log.Error(
			errors.New("more than one unexpected Flink job were found"),
			"", "unexpected jobs", flinkJobsUnexpected)
	}
	if flinkJob != nil {
		log.Info("Observed Flink job", "flink job", *flinkJob)
	}
	return
}
func (observer *ClusterStateObserver) observeSavepoint(observed *ObservedClusterState) error {
	var log = observer.log
	if observed.cluster == nil {
		return nil
	}
	var savepointStatus = observed.cluster.Status.Savepoint
	if savepointStatus != nil && savepointStatus.State == v1beta1.SavepointStateInProgress {
		var flinkAPIBaseURL = getFlinkAPIBaseURL(observed.cluster)
		var jobID = savepointStatus.JobID
		var triggerID = savepointStatus.TriggerID
		var savepoint flinkclient.SavepointStatus
		var err error
		savepoint, err = observer.flinkClient.GetSavepointStatus(flinkAPIBaseURL, jobID, triggerID)
		observed.savepoint = &savepoint
		if err == nil && len(savepoint.FailureCause.StackTrace) > 0 {
			err = fmt.Errorf("%s", savepoint.FailureCause.StackTrace)
		}
		if err != nil {
			observed.savepointErr = err
			log.Info("Failed to get savepoint.", "error", err, "jobID", jobID, "triggerID", triggerID)
		}
		return err
	}
	return nil
}
func (observer *ClusterStateObserver) observeCluster(
	cluster *v1beta1.FlinkCluster) error {
	return observer.k8sClient.Get(
		observer.context, observer.request.NamespacedName, cluster)
}
func (observer *ClusterStateObserver) observeRevisions(
	revisions *[]*appsv1.ControllerRevision,
	cluster *v1beta1.FlinkCluster) error {
	if cluster == nil {
		return nil
	}
	selector := labels.SelectorFromSet(labels.Set(map[string]string{history.ControllerRevisionManagedByLabel: cluster.GetName()}))
	controllerRevisions, err := observer.history.ListControllerRevisions(cluster, selector)
	*revisions = append(*revisions, controllerRevisions...)
	return err
}
func (observer *ClusterStateObserver) observeConfigMap(
	observedConfigMap *corev1.ConfigMap) error {
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	return observer.k8sClient.Get(
		observer.context,
		types.NamespacedName{
			Namespace: clusterNamespace,
			Name:      getConfigMapName(clusterName),
		},
		observedConfigMap)
}
func (observer *ClusterStateObserver) observeJobManagerStatefulSet(
	observedStatefulSet *appsv1.StatefulSet) error {
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	var jmStatefulSetName = getJobManagerStatefulSetName(clusterName)
	return observer.observeStatefulSet(
		clusterNamespace, jmStatefulSetName, "JobManager", observedStatefulSet)
}
func (observer *ClusterStateObserver) observeTaskManagerStatefulSet(
	observedStatefulSet *appsv1.StatefulSet) error {
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	var tmStatefulSetName = getTaskManagerStatefulSetName(clusterName)
	return observer.observeStatefulSet(
		clusterNamespace, tmStatefulSetName, "TaskManager", observedStatefulSet)
}
func (observer *ClusterStateObserver) observeStatefulSet(
	namespace string,
	name string,
	component string,
	observedStatefulSet *appsv1.StatefulSet) error {
	var log = observer.log.WithValues("component", component)
	var err = observer.k8sClient.Get(
		observer.context,
		types.NamespacedName{
			Namespace: namespace,
			Name:      name,
		},
		observedStatefulSet)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error(err, "Failed to get StatefulSet")
		} else {
			log.Info("Deployment not found")
		}
	}
	return err
}
func (observer *ClusterStateObserver) observeJobManagerService(
	observedService *corev1.Service) error {
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	return observer.k8sClient.Get(
		observer.context,
		types.NamespacedName{
			Namespace: clusterNamespace,
			Name:      getJobManagerServiceName(clusterName),
		},
		observedService)
}
func (observer *ClusterStateObserver) observeJobManagerIngress(
	observedIngress *extensionsv1beta1.Ingress) error {
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	return observer.k8sClient.Get(
		observer.context,
		types.NamespacedName{
			Namespace: clusterNamespace,
			Name:      getJobManagerIngressName(clusterName),
		},
		observedIngress)
}
func (observer *ClusterStateObserver) observeJobResource(
	observedJob *batchv1.Job) error {
	var clusterNamespace = observer.request.Namespace
	var clusterName = observer.request.Name
	return observer.k8sClient.Get(
		observer.context,
		types.NamespacedName{
			Namespace: clusterNamespace,
			Name:      getJobName(clusterName),
		},
		observedJob)
}