diff --git a/.gitignore b/.gitignore index 071e3b0..c9c8504 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,5 @@ db *.log tmp *.jar -.env \ No newline at end of file +.env +/operator \ No newline at end of file diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000..f88f353 --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,41 @@ +# Flink Kube Operator + +Kubernetes operator that manages the lifecycle of Apache Flink jobs via CRDs. + +## Language + +**FlinkJob**: +A Kubernetes Custom Resource representing a desired Flink job. Contains spec (desired state) and status (observed state). +_Avoid_: Job CR, CRD instance + +**ManagedJob**: +Runtime wrapper that drives a FlinkJob's lifecycle through its state machine (init → run → savepoint → upgrade). +_Avoid_: Job controller, job runner + +**jarRef**: +Full OCI reference identifying a JAR artifact in a registry (e.g. `registry.io/org/app:1.0.0`). +_Avoid_: jarUri, jar URL, artifact URL + +**jarDigest**: +SHA256 content hash of the JAR layer in the OCI manifest. Immutable identifier for the artifact bytes. +_Avoid_: content hash, checksum + +**pullSecret**: +A `kubernetes.io/dockerconfigjson` Kubernetes Secret containing registry credentials, referenced by name in the FlinkJob spec. +_Avoid_: imagePullSecret, registry secret + +**JarFile**: +Ephemeral artifact handle — a downloaded/pulled JAR on local disk awaiting upload to Flink, then deleted. +_Avoid_: temp jar, local jar + +**LifeCycleStatus**: +Operator-managed lifecycle state of a FlinkJob (HEALTHY, RESTORING, GRACEFULLY_PAUSED, FAILED, etc.). +_Avoid_: operator status, lifecycle state + +**JobStatus**: +Flink-reported job state (RUNNING, FAILED, FINISHED, etc.). Mirrored from the Flink REST API. +_Avoid_: flink status, job state + +**Savepoint**: +A consistent snapshot of a Flink job's state, used for restart and upgrade. +_Avoid_: checkpoint, snapshot diff --git a/README.md b/README.md index a51e71a..bd32ff0 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,203 @@ -Installation: +# Flink Kube Operator + +A Kubernetes operator for managing Apache Flink jobs via Custom Resource Definitions. + +## Installation ```bash helm repo add lc-flink-operator https://git.logicamp.tech/Logicamp/flink-kube-operator/raw/branch/main/helm/ helm install flink-kube-operator lc-flink-operator/flink-kube-operator -``` \ No newline at end of file +``` + +## FlinkJob CRD + +Define Flink jobs as Kubernetes resources: + +```yaml +apiVersion: flink.logicamp.dev/v1alpha1 +kind: FlinkJob +metadata: + name: my-flink-job +spec: + key: word-count + name: "Word Count Example" + entryClass: "com.example.WordCount" + parallelism: 1 + jarRef: "registry.example.com/myorg/word-count:1.0.0" + jarPullSecret: "registry-creds" + args: + kafka-host: kafka:9092 + kafka-group-id: my-group +``` + +### Spec Fields + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `key` | string | yes | Unique identifier for the job | +| `name` | string | yes | Human-readable job name | +| `entryClass` | string | yes | Flink job entry class | +| `parallelism` | int | yes | Job parallelism | +| `jarRef` | string | yes | OCI reference to the JAR artifact (e.g. `registry.io/org/app:tag`) | +| `jarPullSecret` | string | no | Name of a `kubernetes.io/dockerconfigjson` Secret for registry auth | +| `args` | map/list | no | Program arguments (map or list format) | +| `savepointInterval` | duration | no | Interval between automatic savepoints | + +### Deprecated Fields (backward compat) + +| Field | Replaced by | +|-------|-------------| +| `jarUri` | `jarRef` | +| `jarURIBasicAuthUsername` | `jarPullSecret` | +| `jarURIBasicAuthPassword` | `jarPullSecret` | + +If both `jarRef` and `jarUri` are set, `jarRef` takes precedence. + +## Registry Authentication + +### Kubernetes (pull secret) + +Create a `kubernetes.io/dockerconfigjson` Secret and reference it via `jarPullSecret`: + +```bash +kubectl create secret docker-registry registry-creds \ + --docker-server=registry.example.com \ + --docker-username=user \ + --docker-password=pass +``` + +```yaml +spec: + jarRef: "registry.example.com/myorg/my-job:1.0.0" + jarPullSecret: "registry-creds" +``` + +### Docker Compose / local dev (environment variables) + +Set registry credentials as environment variables on the operator container: + +```yaml +# docker-compose.yaml +services: + operator: + image: flink-kube-operator:latest + environment: + REGISTRY_SERVER: "registry.example.com" + REGISTRY_USERNAME: "user" + REGISTRY_PASSWORD: "pass" +``` + +### Credential Resolution Order + +1. `spec.jarPullSecret` — reads the K8s Secret +2. `REGISTRY_USERNAME` + `REGISTRY_PASSWORD` env vars — fallback (optionally scoped by `REGISTRY_SERVER`) +3. Anonymous — public registries + +## Pushing JARs to an OCI Registry + +Flink job JARs are stored as OCI artifacts with media type `application/java-archive`. Use one of the methods below to push your JAR. + +### Using crane (recommended) + +[crane](https://github.com/google/go-containerregistry/tree/main/cmd/crane) is a tool for interacting with OCI registries. + +```bash +# Install crane +go install github.com/google/go-containerregistry/cmd/crane@latest + +# Log in to your registry +crane auth login registry.example.com -u user -p pass + +# Push a JAR as an OCI artifact +crane append \ + --file=my-job.jar \ + --new_tag=registry.example.com/myorg/my-job:1.0.0 \ + --base=scratch +``` + +### Using oras + +[oras](https://oras.land/) is a tool for pushing OCI artifacts. + +```bash +# Install oras +# https://oras.land/docs/installation + +# Log in to your registry +oras login registry.example.com -u user -p pass + +# Push a JAR with the correct media type +oras push registry.example.com/myorg/my-job:1.0.0 \ + --artifact-type application/java-archive \ + my-job.jar:application/java-archive +``` + +### Using Docker (buildkit) + +```bash +# Create a minimal Dockerfile +cat > Dockerfile.jar <<'EOF' +FROM scratch +COPY my-job.jar /my-job.jar +EOF + +# Build and push +docker build -f Dockerfile.jar -t registry.example.com/myorg/my-job:1.0.0 . +docker push registry.example.com/myorg/my-job:1.0.0 +``` + +### GitHub Actions (CI/CD) + +Use [crane](https://github.com/google/go-containerregistry/tree/main/cmd/crane) in your GitHub Actions workflow to push JARs as OCI artifacts. `docker/build-push-action` is not suitable — it builds container images, not arbitrary OCI artifacts. + +```yaml +name: Publish JAR + +on: + push: + tags: + - 'v*' + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }}/my-job + +jobs: + publish: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - uses: actions/checkout@v4 + + - name: Log in to registry + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Set up crane + uses: imjasonh/setup-crane@v0.4 + + - name: Build JAR + run: mvn package -DskipTests + + - name: Push JAR to OCI registry + run: | + crane append \ + --base=scratch \ + --new_tag=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.ref_name }} \ + --new_layer=target/my-job.jar \ + --media_type=application/java-archive +``` + +For GitLab CI, Bitbucket Pipelines, or other CI systems, use the same `crane append` command after authenticating to your registry. + +### Versioning Strategy + +- Use semantic version tags (`:1.0.0`, `:1.1.0`) for explicit version control +- Changing the tag in `spec.jarRef` triggers a job upgrade (pause → re-upload → resume from savepoint) +- The operator stores the SHA256 digest in `status.runningJarDigest` after each pull diff --git a/crds.yaml b/crds.yaml index c4572ad..a272001 100644 --- a/crds.yaml +++ b/crds.yaml @@ -24,7 +24,6 @@ spec: type: object required: - key - - jarUri properties: key: type: string @@ -37,12 +36,21 @@ spec: type: string parallelism: type: integer + jarRef: + type: string + description: "OCI reference to the JAR artifact (e.g. registry.io/org/app:tag). Takes precedence over jarUri." + jarPullSecret: + type: string + description: "Name of a kubernetes.io/dockerconfigjson Secret for registry authentication." jarUri: type: string + description: "Deprecated. Use jarRef instead. HTTP(S) URL to the JAR file." jarURIBasicAuthUsername: type: string + description: "Deprecated. Use jarPullSecret instead." jarURIBasicAuthPassword: type: string + description: "Deprecated. Use jarPullSecret instead." args: description: "Program arguments. Accepts a list of strings (--key, value, ...) or a map of key-value pairs." x-kubernetes-preserve-unknown-fields: true @@ -98,8 +106,15 @@ spec: lastRestoredSavepointRestoredDate: type: string format: time + runningJarDigest: + type: string + description: "SHA256 digest of the currently running JAR layer." + runningJarRef: + type: string + description: "OCI reference of the currently running JAR." runningJarURI: type: string + description: "Deprecated. HTTP URI of the currently running JAR." pauseSavepointTriggerId: type: string restoredCount: diff --git a/example-job.yaml b/example-job.yaml index 6537f48..f202bab 100644 --- a/example-job.yaml +++ b/example-job.yaml @@ -8,9 +8,8 @@ spec: name: "Word Count Example" entryClass: "tech.logicamp.logiline.FacilityEnrichment" parallelism: 1 - jarUri: "https://git.logicamp.tech/api/packages/logiline/generic/facility-enrichment/1.0.0/facility-enrichment.jar" - jarURIBasicAuthUsername: logiline-actrunner - jarURIBasicAuthPassword: daeweeb7ohpaiw3oojiCoong + jarRef: "lcr.logicamp.tech/logiline/facility-enrichment:1.0.0" + jarPullSecret: "lcr-registry-creds" # args accepts two formats: # Map (recommended): args: @@ -24,4 +23,4 @@ spec: # - batch-sales-config-dedup-processor flinkConfiguration: taskmanager.numberOfTaskSlots: "1" - parallelism.default: "1" \ No newline at end of file + parallelism.default: "1" diff --git a/go.mod b/go.mod index dfae126..f325daa 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.23.2 require ( github.com/danielgtaylor/huma/v2 v2.27.0 github.com/gofiber/fiber/v2 v2.52.6 + github.com/google/go-containerregistry v0.20.3 github.com/logi-camp/go-flink-client v0.2.2 github.com/samber/lo v1.47.0 go.uber.org/zap v1.27.0 @@ -14,6 +15,10 @@ require ( require ( github.com/andybalholm/brotli v1.1.1 // indirect + github.com/containerd/stargz-snapshotter/estargz v0.16.3 // indirect + github.com/docker/cli v27.5.0+incompatible // indirect + github.com/docker/distribution v2.8.3+incompatible // indirect + github.com/docker/docker-credential-helpers v0.8.2 // indirect github.com/emicklei/go-restful/v3 v3.11.0 // indirect github.com/evanphx/json-patch/v5 v5.9.0 // indirect github.com/go-openapi/jsonpointer v0.21.0 // indirect @@ -26,13 +31,19 @@ require ( github.com/klauspost/compress v1.17.11 // indirect github.com/mailru/easyjson v0.7.7 // indirect github.com/mattn/go-runewidth v0.0.16 // indirect + github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.0 // indirect github.com/rivo/uniseg v0.4.7 // indirect + github.com/sirupsen/logrus v1.9.3 // indirect github.com/valyala/bytebufferpool v1.0.0 // indirect github.com/valyala/fasthttp v1.58.0 // indirect github.com/valyala/tcplisten v1.0.0 // indirect + github.com/vbatts/tar-split v0.11.6 // indirect golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect - google.golang.org/protobuf v1.35.1 // indirect + golang.org/x/sync v0.10.0 // indirect + google.golang.org/protobuf v1.36.3 // indirect ) require ( @@ -53,7 +64,7 @@ require ( github.com/x448/float16 v0.8.4 // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/net v0.31.0 // indirect - golang.org/x/oauth2 v0.21.0 // indirect + golang.org/x/oauth2 v0.25.0 // indirect golang.org/x/sys v0.29.0 // indirect golang.org/x/term v0.26.0 // indirect golang.org/x/text v0.20.0 // indirect @@ -61,7 +72,7 @@ require ( gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 gopkg.in/yaml.v3 v3.0.1 // indirect - k8s.io/api v0.31.3 // indirect + k8s.io/api v0.31.3 k8s.io/klog/v2 v2.130.1 // indirect k8s.io/kube-openapi v0.0.0-20241127205056-99599406b04f // indirect k8s.io/utils v0.0.0-20241104163129-6fe5fd82f078 // indirect diff --git a/go.sum b/go.sum index 2f856eb..6cec087 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= +github.com/containerd/stargz-snapshotter/estargz v0.16.3 h1:7evrXtoh1mSbGj/pfRccTampEyKpjpOnS3CyiV1Ebr8= +github.com/containerd/stargz-snapshotter/estargz v0.16.3/go.mod h1:uyr4BfYfOj3G9WBVE8cOlQmXAbPN9VEQpBBeJIuOipU= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/danielgtaylor/huma/v2 v2.27.0 h1:yxgJ8GqYqKeXw/EnQ4ZNc2NBpmn49AlhxL2+ksSXjUI= github.com/danielgtaylor/huma/v2 v2.27.0/go.mod h1:NbSFXRoOMh3BVmiLJQ9EbUpnPas7D9BeOxF/pZBAGa0= @@ -7,6 +9,12 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/docker/cli v27.5.0+incompatible h1:aMphQkcGtpHixwwhAXJT1rrK/detk2JIvDaFkLctbGM= +github.com/docker/cli v27.5.0+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8= +github.com/docker/distribution v2.8.3+incompatible h1:AtKxIZ36LoNK51+Z6RpzLpddBirtxJnzDrHLEKxTAYk= +github.com/docker/distribution v2.8.3+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w= +github.com/docker/docker-credential-helpers v0.8.2 h1:bX3YxiGzFP5sOXWc3bTPEXdEaZSeVMrFgOr3T+zrFAo= +github.com/docker/docker-credential-helpers v0.8.2/go.mod h1:P3ci7E3lwkZg6XiHdRKft1KckHiO9a2rNtyFbZ/ry9M= github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g= github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= github.com/evanphx/json-patch/v5 v5.9.0 h1:kcBlZQbplgElYIlo/n1hJbls2z/1awpXxpRi0/FOJfg= @@ -38,6 +46,8 @@ github.com/google/gnostic-models v0.6.9/go.mod h1:CiWsm0s6BSQd1hRn8/QmxqB6BesYcb github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-containerregistry v0.20.3 h1:oNx7IdTI936V8CQRveCjaxOiegWwvM7kqkbXTpyiovI= +github.com/google/go-containerregistry v0.20.3/go.mod h1:w00pIgBRDVUDFM6bq+Qx8lwNWK+cxgCuX1vd3PIBDNI= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0= github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= @@ -73,6 +83,8 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc= github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= +github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y= +github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -84,6 +96,10 @@ github.com/onsi/ginkgo/v2 v2.19.0 h1:9Cnnf7UHo57Hy3k6/m5k3dRfGTMXGvxhHFvkDTCTpvA github.com/onsi/ginkgo/v2 v2.19.0/go.mod h1:rlwLi9PilAFJ8jCg9UE1QP6VBpd6/xj3SRC0d6TU0To= github.com/onsi/gomega v1.33.1 h1:dsYjIxxSR755MDmKVsaFQTE22ChNBcuuTWgkUDSubOk= github.com/onsi/gomega v1.33.1/go.mod h1:U4R44UsT+9eLIaYRB2a5qajjtQYn0hauxvRm16AVYg0= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug= +github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= @@ -96,12 +112,15 @@ github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= github.com/samber/lo v1.47.0 h1:z7RynLwP5nbyRscyvcD043DWYoOcYRv3mV8lBeqOCLc= github.com/samber/lo v1.47.0/go.mod h1:RmDH9Ct32Qy3gduHQuKJ3gW1fMHAnE/fAzQuf6He5cU= +github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= +github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= @@ -113,6 +132,8 @@ github.com/valyala/fasthttp v1.58.0 h1:GGB2dWxSbEprU9j0iMJHgdKYJVDyjrOwF9RE59PbR github.com/valyala/fasthttp v1.58.0/go.mod h1:SYXvHHaFp7QZHGKSHmoMipInhrI5StHrhDTYVEjK/Kw= github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8= github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc= +github.com/vbatts/tar-split v0.11.6 h1:4SjTW5+PU11n6fZenf2IPoV8/tz3AaYHMWjf23envGs= +github.com/vbatts/tar-split v0.11.6/go.mod h1:dqKNtesIOr2j2Qv3W/cHjnvk9I8+G7oAkFDFN6TCBEI= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= @@ -140,12 +161,17 @@ golang.org/x/net v0.31.0 h1:68CPQngjLL0r2AlUKiSxtQFKvzRVbnzLwMUn5SzcLHo= golang.org/x/net v0.31.0/go.mod h1:P4fl1q7dY2hnZFxEk4pPSkDHF+QqjitcnDjUQyMM+pM= golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs= golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI= +golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70= +golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= +golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU= @@ -164,12 +190,15 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.24.0 h1:J1shsA93PJUEVaUSaay7UXAyE8aimq3GW0pjlolpa24= golang.org/x/tools v0.24.0/go.mod h1:YhNqVBIfWHdzvTLs0d8LCuMhkKUgSUKldakyV7W/WDQ= +golang.org/x/tools v0.29.0 h1:Xx0h3TtM9rzQpQuR4dKLrdglAmCEN5Oi+P74JdhdzXE= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA= google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= +google.golang.org/protobuf v1.36.3 h1:82DV7MYdb8anAVi3qge1wSnMDrnKK7ebr+I0hHRN1BU= +google.golang.org/protobuf v1.36.3/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/internal/crd/auth.go b/internal/crd/auth.go new file mode 100644 index 0000000..91e8d5e --- /dev/null +++ b/internal/crd/auth.go @@ -0,0 +1,83 @@ +package crd + +import ( + "context" + "encoding/json" + "fmt" + "os" + + "github.com/google/go-containerregistry/pkg/authn" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" +) + +type dockerConfigJSON struct { + Auths map[string]authn.AuthConfig `json:"auths"` +} + +// ResolveAuth returns an OCI authenticator for the given registry. +// It checks in order: pull secret from K8s, then environment variables. +// Returns authn.Anonymous if no credentials are found. +func (crd *Crd) ResolveAuth(pullSecretName string, server string) (authn.Authenticator, error) { + if pullSecretName != "" { + return crd.authFromPullSecret(pullSecretName, server) + } + return authFromEnv(server), nil +} + +func (crd *Crd) authFromPullSecret(secretName string, server string) (authn.Authenticator, error) { + ns := os.Getenv("NAMESPACE") + secret := &corev1.Secret{} + err := crd.runtimeClient.Get(context.Background(), types.NamespacedName{ + Namespace: ns, + Name: secretName, + }, secret) + if err != nil { + return nil, fmt.Errorf("reading pull secret %q: %w", secretName, err) + } + + if secret.Type != corev1.SecretTypeDockerConfigJson { + return nil, fmt.Errorf("secret %q is not of type kubernetes.io/dockerconfigjson", secretName) + } + + dockerConfigJSONBytes, ok := secret.Data[corev1.DockerConfigJsonKey] + if !ok { + return nil, fmt.Errorf("secret %q missing .dockerconfigjson key", secretName) + } + + var cfg dockerConfigJSON + if err := json.Unmarshal(dockerConfigJSONBytes, &cfg); err != nil { + return nil, fmt.Errorf("parsing docker config from secret %q: %w", secretName, err) + } + + if server != "" { + if entry, ok := cfg.Auths[server]; ok { + return authn.FromConfig(entry), nil + } + } + + for _, entry := range cfg.Auths { + return authn.FromConfig(entry), nil + } + + return authn.Anonymous, nil +} + +func authFromEnv(server string) authn.Authenticator { + username := os.Getenv("REGISTRY_USERNAME") + password := os.Getenv("REGISTRY_PASSWORD") + envServer := os.Getenv("REGISTRY_SERVER") + + if username == "" || password == "" { + return authn.Anonymous + } + + if envServer != "" && server != "" && envServer != server { + return authn.Anonymous + } + + return authn.FromConfig(authn.AuthConfig{ + Username: username, + Password: password, + }) +} diff --git a/internal/crd/v1alpha1/flink_job.go b/internal/crd/v1alpha1/flink_job.go index a920710..5cba344 100644 --- a/internal/crd/v1alpha1/flink_job.go +++ b/internal/crd/v1alpha1/flink_job.go @@ -60,16 +60,22 @@ func (a Args) MarshalJSON() ([]byte, error) { } 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"` + 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 { @@ -85,7 +91,10 @@ type FlinkJobStatus struct { LastRestoredSavepointDate *time.Time `json:"lastRestoredSavepointDate,omitempty"` LastRestoredSavepointRestoredDate *time.Time `json:"lastRestoredSavepointRestoredDate,omitempty"` RestoredCount int `json:"restoredCount,omitempty"` - RunningJarURI *string `json:"runningJarURI"` + 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 @@ -140,3 +149,48 @@ const ( 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 +} diff --git a/internal/crd/v1alpha1/zz_generated.deepcopy.go b/internal/crd/v1alpha1/zz_generated.deepcopy.go index 7e1529e..c2caa6a 100644 --- a/internal/crd/v1alpha1/zz_generated.deepcopy.go +++ b/internal/crd/v1alpha1/zz_generated.deepcopy.go @@ -5,6 +5,8 @@ package v1alpha1 import ( + "time" + runtime "k8s.io/apimachinery/pkg/runtime" ) @@ -13,7 +15,8 @@ func (in *FlinkJob) DeepCopyInto(out *FlinkJob) { *out = *in out.TypeMeta = in.TypeMeta in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) - out.Spec = in.Spec + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlinkJob. @@ -34,6 +37,111 @@ func (in *FlinkJob) DeepCopyObject() runtime.Object { return nil } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *FlinkJobSpec) DeepCopyInto(out *FlinkJobSpec) { + *out = *in + if in.JarURIBasicAuthUsername != nil { + in, out := &in.JarURIBasicAuthUsername, &out.JarURIBasicAuthUsername + *out = new(string) + **out = **in + } + if in.JarURIBasicAuthPassword != nil { + in, out := &in.JarURIBasicAuthPassword, &out.JarURIBasicAuthPassword + *out = new(string) + **out = **in + } + if in.Args != nil { + in, out := &in.Args, &out.Args + *out = make(Args, len(*in)) + copy(*out, *in) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlinkJobSpec. +func (in *FlinkJobSpec) DeepCopy() *FlinkJobSpec { + if in == nil { + return nil + } + out := new(FlinkJobSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *FlinkJobStatus) DeepCopyInto(out *FlinkJobStatus) { + *out = *in + if in.LastSavepointPath != nil { + in, out := &in.LastSavepointPath, &out.LastSavepointPath + *out = new(string) + **out = **in + } + if in.JarId != nil { + in, out := &in.JarId, &out.JarId + *out = new(string) + **out = **in + } + if in.JobId != nil { + in, out := &in.JobId, &out.JobId + *out = new(string) + **out = **in + } + if in.Error != nil { + in, out := &in.Error, &out.Error + *out = new(string) + **out = **in + } + if in.SavepointTriggerId != nil { + in, out := &in.SavepointTriggerId, &out.SavepointTriggerId + *out = new(string) + **out = **in + } + if in.PauseSavepointTriggerId != nil { + in, out := &in.PauseSavepointTriggerId, &out.PauseSavepointTriggerId + *out = new(string) + **out = **in + } + if in.LastSavepointDate != nil { + in, out := &in.LastSavepointDate, &out.LastSavepointDate + *out = new(time.Time) + **out = **in + } + if in.LastRestoredSavepointDate != nil { + in, out := &in.LastRestoredSavepointDate, &out.LastRestoredSavepointDate + *out = new(time.Time) + **out = **in + } + if in.LastRestoredSavepointRestoredDate != nil { + in, out := &in.LastRestoredSavepointRestoredDate, &out.LastRestoredSavepointRestoredDate + *out = new(time.Time) + **out = **in + } + if in.RunningJarDigest != nil { + in, out := &in.RunningJarDigest, &out.RunningJarDigest + *out = new(string) + **out = **in + } + if in.RunningJarRef != nil { + in, out := &in.RunningJarRef, &out.RunningJarRef + *out = new(string) + **out = **in + } + if in.RunningJarURI != nil { + in, out := &in.RunningJarURI, &out.RunningJarURI + *out = new(string) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlinkJobStatus. +func (in *FlinkJobStatus) DeepCopy() *FlinkJobStatus { + if in == nil { + return nil + } + out := new(FlinkJobStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *FlinkJobList) DeepCopyInto(out *FlinkJobList) { *out = *in diff --git a/internal/jar/jar.go b/internal/jar/jar.go index 78de3d4..9264088 100644 --- a/internal/jar/jar.go +++ b/internal/jar/jar.go @@ -2,31 +2,51 @@ package jar import ( "crypto/rand" - "encoding/base64" "encoding/hex" "errors" + "fmt" "io" "net/http" - "net/http/cookiejar" "os" "strings" "flink-kube-operator/pkg" + "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/go-containerregistry/pkg/name" + "github.com/google/go-containerregistry/pkg/v1/remote" + "github.com/google/go-containerregistry/pkg/v1/types" api "github.com/logi-camp/go-flink-client" "go.uber.org/zap" ) type JarFile struct { - uri string + ref string filePath string + digest string basicAuthUsername *string basicAuthPassword *string + auth authn.Authenticator } -func NewJarFile(URI string, basicAuthUsername *string, basicAuthPassword *string) (*JarFile, error) { +// NewJarFile creates a JarFile from an OCI reference, pulling the artifact. +// auth is the OCI authenticator (from pull secret or env vars). Use authn.Anonymous for public registries. +func NewJarFile(ref string, auth authn.Authenticator) (*JarFile, error) { jarFile := &JarFile{ - uri: URI, + ref: ref, + auth: auth, + } + err := jarFile.Pull() + if err != nil { + return nil, err + } + return jarFile, nil +} + +// NewJarFileFromURI creates a JarFile from an HTTP URI (deprecated path, kept for backward compat). +func NewJarFileFromURI(uri string, basicAuthUsername *string, basicAuthPassword *string) (*JarFile, error) { + jarFile := &JarFile{ + ref: uri, basicAuthUsername: basicAuthUsername, basicAuthPassword: basicAuthPassword, } @@ -38,10 +58,10 @@ func NewJarFile(URI string, basicAuthUsername *string, basicAuthPassword *string } func (jarFile *JarFile) Upload(flinkClient *api.Client) (fileName string, err error) { - resp, err := flinkClient.UploadJar(jarFile.filePath) if err != nil { - pkg.Logger.Error("[main] error uploading jar", zap.Error(err)) + pkg.Logger.Error("[jar] error uploading jar", zap.Error(err)) + return "", err } filePathParts := strings.Split(resp.FileName, "/") fileName = filePathParts[len(filePathParts)-1] @@ -52,6 +72,78 @@ func (jarFile *JarFile) Upload(flinkClient *api.Client) (fileName string, err er return } +func (jarFile *JarFile) Digest() string { + return jarFile.digest +} + +func (jarFile *JarFile) Pull() error { + randBytes := make([]byte, 16) + rand.Read(randBytes) + fileName := hex.EncodeToString(randBytes) + jarFile.filePath = "/tmp/" + fileName + ".jar" + + ref, err := name.ParseReference(jarFile.ref) + if err != nil { + return fmt.Errorf("parsing OCI reference %q: %w", jarFile.ref, err) + } + + img, err := remote.Image(ref, remote.WithAuth(jarFile.auth)) + if err != nil { + return fmt.Errorf("fetching OCI image %q: %w", jarFile.ref, err) + } + + d, err := img.Digest() + if err != nil { + return fmt.Errorf("getting image digest: %w", err) + } + jarFile.digest = d.String() + + layers, err := img.Layers() + if err != nil { + return fmt.Errorf("getting image layers: %w", err) + } + if len(layers) == 0 { + return fmt.Errorf("OCI image %q has no layers", jarFile.ref) + } + + layer := layers[0] + mediaType, err := layer.MediaType() + if err != nil { + return fmt.Errorf("getting layer media type: %w", err) + } + if mediaType != types.OCILayer && mediaType != types.OCIUncompressedLayer && mediaType != types.DockerLayer && mediaType != types.DockerUncompressedLayer && string(mediaType) != "application/java-archive" { + pkg.Logger.Warn("[jar] unexpected layer media type, expected application/java-archive or standard OCI layer", + zap.String("got", string(mediaType)), + zap.String("ref", jarFile.ref), + ) + } + + blob, err := layer.Compressed() + if err != nil { + return fmt.Errorf("reading layer blob: %w", err) + } + defer blob.Close() + + out, err := os.Create(jarFile.filePath) + if err != nil { + return fmt.Errorf("creating temp file: %w", err) + } + defer out.Close() + + _, err = io.Copy(out, blob) + if err != nil { + jarFile.delete() + return fmt.Errorf("writing layer to disk: %w", err) + } + + pkg.Logger.Info("[jar] pulled OCI artifact", + zap.String("ref", jarFile.ref), + zap.String("digest", jarFile.digest), + zap.String("path", jarFile.filePath), + ) + return nil +} + func (jarFile *JarFile) Download() error { randBytes := make([]byte, 16) rand.Read(randBytes) @@ -66,31 +158,15 @@ func (jarFile *JarFile) Download() error { var resp *http.Response if jarFile.basicAuthPassword != nil && jarFile.basicAuthUsername != nil { - - basicAuth := func(username, password string) string { - auth := username + ":" + password - return base64.StdEncoding.EncodeToString([]byte(auth)) - } - - redirectPolicyFunc := func(req *http.Request, via []*http.Request) error { - req.Header.Add("Authorization", "Basic "+basicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword)) - return nil - } - - client := &http.Client{ - Jar: &cookiejar.Jar{}, - CheckRedirect: redirectPolicyFunc, - } - - req, err := http.NewRequest("GET", jarFile.uri, nil) + req, err := http.NewRequest("GET", jarFile.ref, nil) if err != nil { jarFile.delete() return err } - req.Header.Add("Authorization", "Basic "+basicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword)) - resp, err = client.Do(req) + req.SetBasicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword) + resp, err = http.DefaultClient.Do(req) } else { - resp, err = http.Get(jarFile.uri) + resp, err = http.Get(jarFile.ref) } if err != nil { jarFile.delete() @@ -98,9 +174,7 @@ func (jarFile *JarFile) Download() error { return err } if resp.StatusCode > 299 { - respBody := []byte{} - resp.Body.Read(respBody) - err = errors.New(string(respBody) + " status:" + resp.Status) + err = fmt.Errorf("download failed: status %s", resp.Status) pkg.Logger.Error("error in downloading jar", zap.Error(err)) return err } diff --git a/internal/managed_job/cycle.go b/internal/managed_job/cycle.go index 157e168..73e9538 100644 --- a/internal/managed_job/cycle.go +++ b/internal/managed_job/cycle.go @@ -10,8 +10,6 @@ import ( ) func (job *ManagedJob) Cycle() { - // pkg.Logger.Debug("[managed-job] [new] check cycle", zap.String("jobName", job.def.GetName())) - // Init job if job.def.Status.LifeCycleStatus == "" && (job.def.Status.JobStatus == "" || job.def.Status.JobStatus == v1alpha1.JobStatusFinished) { job.Run(false) @@ -26,7 +24,8 @@ func (job *ManagedJob) Cycle() { job.trackSavepoint() } } - if job.def.Status.RunningJarURI != nil && job.def.Spec.JarURI != *job.def.Status.RunningJarURI { + + if job.def.Spec.NeedsUpgrade(job.def.Status) { job.upgrade() } @@ -42,14 +41,6 @@ func (job *ManagedJob) Cycle() { job.crd.SetJobStatus(job.def.UID, job.def.Status) return } - // if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath != nil { - // //job.restore() - // return - // } - // if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath == nil { - // //job.restore() - // return - // } pkg.Logger.Warn("[managed-job] [cycle] unhandled job status", zap.String("name", job.def.Name), zap.String("status", string(job.def.Status.JobStatus)), zap.String("namespace", job.def.Namespace)) } diff --git a/internal/managed_job/jar.go b/internal/managed_job/jar.go index 0f376b4..c4b7fa1 100644 --- a/internal/managed_job/jar.go +++ b/internal/managed_job/jar.go @@ -4,41 +4,90 @@ import ( "flink-kube-operator/internal/jar" "flink-kube-operator/pkg" + "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/go-containerregistry/pkg/name" "go.uber.org/zap" ) -// upload jar file and set the jarId for later usages +// upload pulls/downloads the jar and uploads it to Flink, then patches the CRD status. func (job *ManagedJob) upload() error { - jarFile, err := jar.NewJarFile(job.def.Spec.JarURI, job.def.Spec.JarURIBasicAuthUsername, job.def.Spec.JarURIBasicAuthPassword) + spec := job.def.Spec + var jarFile *jar.JarFile + var err error + + if spec.IsOCISource() { + auth, authErr := job.resolveAuth() + if authErr != nil { + pkg.Logger.Error("[managed-job] [upload] auth resolution failed", zap.Error(authErr)) + return authErr + } + jarFile, err = jar.NewJarFile(spec.EffectiveJarRef(), auth) + } else { + jarFile, err = jar.NewJarFileFromURI(spec.JarURI, spec.JarURIBasicAuthUsername, spec.JarURIBasicAuthPassword) + } + if err != nil { - pkg.Logger.Debug("[manage-job] [upload] error on download jar", zap.Error(err)) + pkg.Logger.Debug("[managed-job] [upload] error on pull/download jar", zap.Error(err)) return err } + jarId, err := jarFile.Upload(job.client) if err != nil { - pkg.Logger.Debug("[manage-job] [upload] error on upload jar", zap.Error(err)) + pkg.Logger.Debug("[managed-job] [upload] error on upload jar", zap.Error(err)) return err } - pkg.Logger.Info("[manage-job] [upload] uploaded", zap.Any("upload-jar-resp", jarId)) + pkg.Logger.Info("[managed-job] [upload] uploaded", zap.String("jarId", jarId)) - job.def.Status.JarId = &jarId - job.crd.Patch(job.def.UID, map[string]interface{}{ + patchData := map[string]interface{}{ "status": map[string]interface{}{ - "jarId": job.def.Status.JarId, + "jarId": &jarId, }, - }) + } + + if spec.IsOCISource() { + digest := jarFile.Digest() + patchData["status"].(map[string]interface{})["runningJarDigest"] = &digest + } + + job.crd.Patch(job.def.UID, patchData) return nil } +func (job *ManagedJob) resolveAuth() (authn.Authenticator, error) { + s := job.def.Spec + server := "" + ref := s.EffectiveJarRef() + if reg, err := extractRegistry(ref); err == nil { + server = reg + } + return job.crd.ResolveAuth(s.JarPullSecret, server) +} + +func extractRegistry(ref string) (string, error) { + r, err := name.NewRegistry(ref) + if err != nil { + t, err2 := name.NewTag(ref) + if err2 != nil { + return "", err + } + return t.RegistryStr(), nil + } + return r.RegistryStr(), nil +} + func (job *ManagedJob) RemoveJar() { if job.def.Status.JarId != nil { err := job.client.DeleteJar(*job.def.Status.JarId) - pkg.Logger.Error("[managed-job] [jar]", zap.Error(err)) + if err != nil { + pkg.Logger.Error("[managed-job] [jar] delete jar error", zap.Error(err)) + } err = job.crd.Patch(job.def.UID, map[string]interface{}{ "status": map[string]interface{}{ "jarId": nil, }, }) - pkg.Logger.Error("[managed-job] [jar]", zap.Error(err)) + if err != nil { + pkg.Logger.Error("[managed-job] [jar] patch jarId error", zap.Error(err)) + } } } diff --git a/internal/managed_job/run.go b/internal/managed_job/run.go index 5aa48f1..06af0a2 100644 --- a/internal/managed_job/run.go +++ b/internal/managed_job/run.go @@ -83,17 +83,21 @@ func (job *ManagedJob) Run(restoreMode bool) error { // job.def.Status.JobId = &runJarResp.JobId // job.def.Status.JobStatus = v1alpha1.JobStatusCreating // job.def.Status.Error = nil + statusPatch := map[string]interface{}{ + "jobId": jobId, + "jobStatus": v1alpha1.JobStatusCreating, + "lifeCycleStatus": v1alpha1.LifeCycleStatusRestoring, + "lastRestoredSavepointDate": job.def.Status.LastSavepointDate, + "restoredCount": job.def.Status.RestoredCount + 1, + "lastRestoredSavepointRestoredDate": time.Now().Format(time.RFC3339), + "error": nil, + } + + refKey, refVal := job.def.Spec.RunningRefPatchData() + statusPatch[refKey] = refVal + job.crd.Patch(job.def.UID, map[string]interface{}{ - "status": map[string]interface{}{ - "jobId": jobId, - "runningJarURI": job.def.Spec.JarURI, - "jobStatus": v1alpha1.JobStatusCreating, - "lifeCycleStatus": v1alpha1.LifeCycleStatusRestoring, - "lastRestoredSavepointDate": job.def.Status.LastSavepointDate, - "restoredCount": job.def.Status.RestoredCount + 1, - "lastRestoredSavepointRestoredDate": time.Now().Format(time.RFC3339), - "error": nil, - }, + "status": statusPatch, }) return nil diff --git a/internal/managed_job/upgrade.go b/internal/managed_job/upgrade.go index bf16fbf..40f23da 100644 --- a/internal/managed_job/upgrade.go +++ b/internal/managed_job/upgrade.go @@ -7,10 +7,13 @@ import ( ) func (job *ManagedJob) upgrade() { - pkg.Logger.Info("[managed-job] [upgrade] pausing... ", + effectiveRef := job.def.Spec.EffectiveJarRef() + prevRef := job.def.Status.RunningRef() + + pkg.Logger.Info("[managed-job] [upgrade] pausing...", zap.String("jobName", job.def.GetName()), - zap.String("currentJarURI", job.def.Spec.JarURI), - zap.String("prevJarURI", *job.def.Status.RunningJarURI), + zap.String("currentRef", effectiveRef), + zap.String("prevRef", prevRef), ) job.def.Status.JarId = nil job.crd.Patch(job.def.UID, map[string]interface{}{ @@ -23,11 +26,10 @@ func (job *ManagedJob) upgrade() { pkg.Logger.Error("[managed-job] [upgrade] error in pausing", zap.Error(err)) return } - pkg.Logger.Info("[managed-job] [upgrade] restoring... ", + pkg.Logger.Info("[managed-job] [upgrade] restoring...", zap.String("jobName", job.def.GetName()), - zap.String("currentJarURI", job.def.Spec.JarURI), - zap.String("prevJarURI", *job.def.Status.RunningJarURI), - zap.Error(err), + zap.String("currentRef", effectiveRef), + zap.String("prevRef", prevRef), ) err = job.Run(true)