flink-kube-operator/internal/crd/v1alpha1/flink_job.go

package v1alpha1

import (
	"encoding/json"
	"errors"
	"sort"
	"strings"
	"time"

	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

//go:generate go run sigs.k8s.io/controller-tools/cmd/controller-gen object paths=$GOFILE

// Args is a list of program arguments passed to the Flink job.
// It accepts two formats in JSON/YAML:
//
//	List:  ["--kafka-host", "broker:9092", "--kafka-group-id", "my-group"]
//	Map:   {kafka-host: "broker:9092", kafka-group-id: "my-group"}
//
// Map keys are automatically prefixed with "--" and expanded into alternating
// key/value pairs. The internal representation is always a flat []string.
type Args []string

// UnmarshalJSON implements dual-format deserialization.
func (a *Args) UnmarshalJSON(data []byte) error {
	// Try map format first: {"key": "value", ...}
	var m map[string]string
	if err := json.Unmarshal(data, &m); err == nil {
		*a = make(Args, 0, len(m)*2)
		// Sort keys for deterministic ordering.
		keys := make([]string, 0, len(m))
		for k := range m {
			keys = append(keys, k)
		}
		sort.Strings(keys)
		for _, k := range keys {
			*a = append(*a, "--"+k, m[k])
		}
		return nil
	}

	// Fall back to list format: ["--key", "value", ...]
	var s []string
	if err := json.Unmarshal(data, &s); err != nil {
		return err
	}
	*a = Args(s)
	return nil
}

// MarshalJSON serializes Args in the compact map format.
func (a Args) MarshalJSON() ([]byte, error) {
	m := make(map[string]string, len(a)/2)
	for i := 0; i < len(a)-1; i += 2 {
		key := strings.TrimPrefix(a[i], "--")
		m[key] = a[i+1]
	}
	return json.Marshal(m)
}

type FlinkJobSpec struct {
	Key              string          `json:"key"`
	Name             string          `json:"name"`
	FlinkCluster     string          `json:"flinkCluster"`
	Parallelism      int             `json:"parallelism"`
	JarRef           string          `json:"jarRef,omitempty"`
	JarPullSecret    string          `json:"jarPullSecret,omitempty"`
	SavepointInterval metaV1.Duration `json:"savepointInterval"`
	EntryClass       string          `json:"entryClass"`
	Args             Args            `json:"args,omitempty"`

	// Deprecated: use JarRef instead. Kept for backward compatibility.
	JarURI string `json:"jarUri,omitempty"`
	// Deprecated: use JarPullSecret instead. Kept for backward compatibility.
	JarURIBasicAuthUsername *string `json:"jarURIBasicAuthUsername,omitempty"`
	// Deprecated: use JarPullSecret instead. Kept for backward compatibility.
	JarURIBasicAuthPassword *string `json:"jarURIBasicAuthPassword,omitempty"`
}

type FlinkJobStatus struct {
	JobStatus                         JobStatus       `json:"jobStatus,omitempty"`
	LifeCycleStatus                   LifeCycleStatus `json:"lifeCycleStatus,omitempty"`
	LastSavepointPath                 *string         `json:"lastSavepointPath,omitempty"`
	JarId                             *string         `json:"jarId,omitempty"`
	JobId                             *string         `json:"jobId,omitempty"`
	Error                             *string         `json:"error,omitempty"`
	SavepointTriggerId                *string         `json:"savepointTriggerId,omitempty"`
	PauseSavepointTriggerId           *string         `json:"pauseSavepointTriggerId,omitempty"`
	LastSavepointDate                 *time.Time      `json:"lastSavepointDate,omitempty"`
	LastRestoredSavepointDate         *time.Time      `json:"lastRestoredSavepointDate,omitempty"`
	LastRestoredSavepointRestoredDate *time.Time      `json:"lastRestoredSavepointRestoredDate,omitempty"`
	RestoredCount                     int             `json:"restoredCount,omitempty"`
	RunningJarDigest                  *string         `json:"runningJarDigest,omitempty"`
	RunningJarRef                     *string         `json:"runningJarRef,omitempty"`
	// Deprecated: kept for backward compatibility during migration.
	RunningJarURI *string `json:"runningJarURI,omitempty"`
}

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
type FlinkJob struct {
	metaV1.TypeMeta   `json:",inline"`
	metaV1.ObjectMeta `json:"metadata,omitempty"`
	Spec              FlinkJobSpec   `json:"spec"`
	Status            FlinkJobStatus `json:"status"`
}

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
type FlinkJobList struct {
	metaV1.TypeMeta `json:",inline"`
	metaV1.ListMeta `json:"metadata,omitempty"`
	Items           []FlinkJob `json:"items"`
}

var (
	ErrNoJobId              = errors.New("[managed-job] no job id")
	ErrNoJarId              = errors.New("[managed-job] no jar id")
	ErrNoSavepointTriggerId = errors.New("[managed-job] no savepoint trigger id")
	ErrNoSavepointPath      = errors.New("[managed-job] no savepoint path")
	ErrOnStartingJob        = errors.New("[managed-job] error on starting job")
)

type JobStatus string

var (
	JobStatusInitializing JobStatus = "INITIALIZING"
	JobStatusRunning      JobStatus = "RUNNING"
	JobStatusCreating     JobStatus = "CREATING"
	JobStatusError        JobStatus = "ERROR"
	JobStatusReconciling  JobStatus = "RECONCILING"
	JobStatusFailed       JobStatus = "FAILED"
	JobStatusFailing      JobStatus = "FAILING"
	JobStatusRestarting   JobStatus = "RESTARTING"
	JobStatusFinished     JobStatus = "FINISHED"
	JobStatusCanceled     JobStatus = "CANCELED"
	JobStatusCancelling   JobStatus = "CANCELLING"
	JobStatusSuspended    JobStatus = "SUSPENDED"
)

type LifeCycleStatus string

const (
	LifeCycleStatusInitializing        LifeCycleStatus = "INITIALIZING"
	LifeCycleStatusRestoring           LifeCycleStatus = "RESTORING"
	LifeCycleStatusGracefulStopFailed  LifeCycleStatus = "GRACEFUL_STOP_FAILED"
	LifeCycleStatusUpgradeFailed       LifeCycleStatus = "UPGRADE_FAILED"
	LifeCycleStatusGracefullyPaused    LifeCycleStatus = "GRACEFULLY_PAUSED"
	LifeCycleStatusUnhealthyJobManager LifeCycleStatus = "UNHEALTHY_JOB_MANAGER"
	LifeCycleStatusHealthy             LifeCycleStatus = "HEALTHY"
	LifeCycleStatusFailed              LifeCycleStatus = "FAILED"
)

// EffectiveJarRef returns the OCI reference, preferring JarRef over the deprecated JarURI.
func (s *FlinkJobSpec) EffectiveJarRef() string {
	if s.JarRef != "" {
		return s.JarRef
	}
	return s.JarURI
}

// IsOCISource returns true if the job uses an OCI reference (jarRef) rather than an HTTP URL (jarUri).
func (s *FlinkJobSpec) IsOCISource() bool {
	return s.JarRef != ""
}

// RunningRef returns the currently tracked running jar reference,
// preferring RunningJarRef (OCI) over the deprecated RunningJarURI.
func (st *FlinkJobStatus) RunningRef() string {
	if st.RunningJarRef != nil {
		return *st.RunningJarRef
	}
	if st.RunningJarURI != nil {
		return *st.RunningJarURI
	}
	return ""
}

// NeedsUpgrade returns true if the spec's jar reference differs from
// what is currently running, indicating an upgrade is needed.
func (s *FlinkJobSpec) NeedsUpgrade(st FlinkJobStatus) bool {
	running := st.RunningRef()
	if running == "" {
		return false
	}
	return s.EffectiveJarRef() != running
}

// RunningRefPatchKey returns the status patch key and value for tracking
// which jar reference is currently running.
func (s *FlinkJobSpec) RunningRefPatchData() (key string, value interface{}) {
	if s.IsOCISource() {
		ref := s.EffectiveJarRef()
		return "runningJarRef", &ref
	}
	return "runningJarURI", s.JarURI
}