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"`
	JarURI                  string          `json:"jarUri"`
	JarURIBasicAuthUsername *string         `json:"jarURIBasicAuthUsername"`
	JarURIBasicAuthPassword *string         `json:"jarURIBasicAuthPassword"`
	SavepointInterval       metaV1.Duration `json:"savepointInterval"`
	EntryClass              string          `json:"entryClass"`
	Args                    Args            `json:"args,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"`
	RunningJarURI                     *string         `json:"runningJarURI"`
}

// +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"
)