mirror of
https://github.com/lukaszraczylo/jobs-manager-operator.git
synced 2026-10-12 05:32:52 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c99558314c | ||
|
|
56edfc8d8c
|
||
|
|
5c92083161
|
||
|
|
60cb552179
|
||
|
|
9d3086c839
|
||
|
|
6189a2e2f3
|
||
|
|
6e39505910
|
||
|
|
9ab900dbd2
|
||
|
|
55cd8bb7a6 | ||
|
|
5fe70085ff | ||
|
|
2133a0e720
|
||
|
|
f87a841746
|
||
|
|
a10fb70e7c
|
||
|
|
3c4f7a56c0
|
||
|
|
55a9a76135
|
||
|
|
ffdc5a1d55
|
||
|
|
90d04588f8
|
||
|
|
262a20d30f
|
||
|
|
e445f55807
|
@@ -100,6 +100,7 @@ jobs:
|
||||
name: "Building docker image"
|
||||
needs: [ prepare, test ]
|
||||
runs-on: ubuntu-20.04
|
||||
if: needs.prepare.outputs.CHECK_IF_MASTER_BRANCH == 'true'
|
||||
steps:
|
||||
- name: Checkout repo
|
||||
uses: actions/checkout@v3
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
# For most projects, this workflow file will not need changing; you simply need
|
||||
# to commit it to your repository.
|
||||
#
|
||||
# You may wish to alter this file to override the set of languages analyzed,
|
||||
# or to provide custom queries or build logic.
|
||||
#
|
||||
# ******** NOTE ********
|
||||
# We have attempted to detect the languages in your repository. Please check
|
||||
# the `language` matrix defined below to confirm you have the correct set of
|
||||
# supported CodeQL languages.
|
||||
#
|
||||
name: "CodeQL"
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ "main" ]
|
||||
pull_request:
|
||||
# The branches below must be a subset of the branches above
|
||||
branches: [ "main" ]
|
||||
schedule:
|
||||
- cron: '22 9 * * 0'
|
||||
|
||||
jobs:
|
||||
analyze:
|
||||
name: Analyze
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
actions: read
|
||||
contents: read
|
||||
security-events: write
|
||||
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
language: [ 'go' ]
|
||||
# CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
|
||||
# Use only 'java' to analyze code written in Java, Kotlin or both
|
||||
# Use only 'javascript' to analyze code written in JavaScript, TypeScript or both
|
||||
# Learn more about CodeQL language support at https://aka.ms/codeql-docs/language-support
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v3
|
||||
|
||||
# Initializes the CodeQL tools for scanning.
|
||||
- name: Initialize CodeQL
|
||||
uses: github/codeql-action/init@v2
|
||||
with:
|
||||
languages: ${{ matrix.language }}
|
||||
# If you wish to specify custom queries, you can do so here or in a config file.
|
||||
# By default, queries listed here will override any specified in a config file.
|
||||
# Prefix the list here with "+" to use these queries and those in the config file.
|
||||
|
||||
# Details on CodeQL's query packs refer to : https://docs.github.com/en/code-security/code-scanning/automatically-scanning-your-code-for-vulnerabilities-and-errors/configuring-code-scanning#using-queries-in-ql-packs
|
||||
# queries: security-extended,security-and-quality
|
||||
|
||||
|
||||
# Autobuild attempts to build any compiled languages (C/C++, C#, Go, or Java).
|
||||
# If this step fails, then you should remove it and run the build manually (see below)
|
||||
- name: Autobuild
|
||||
uses: github/codeql-action/autobuild@v2
|
||||
|
||||
# ℹ️ Command-line programs to run using the OS shell.
|
||||
# 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
|
||||
|
||||
# If the Autobuild fails above, remove it and uncomment the following three lines.
|
||||
# modify them (or add more) to build your code if your project, please refer to the EXAMPLE below for guidance.
|
||||
|
||||
# - run: |
|
||||
# echo "Run, Build Application using script"
|
||||
# ./location_of_script_within_repo/buildscript.sh
|
||||
|
||||
- name: Perform CodeQL Analysis
|
||||
uses: github/codeql-action/analyze@v2
|
||||
with:
|
||||
category: "/language:${{matrix.language}}"
|
||||
@@ -0,0 +1 @@
|
||||
* lukaszraczylo
|
||||
@@ -28,7 +28,6 @@ metadata:
|
||||
labels:
|
||||
name: managedjob-sample
|
||||
spec:
|
||||
# Globally defined parameters and environment variables
|
||||
retries: 3
|
||||
params:
|
||||
env:
|
||||
@@ -40,9 +39,7 @@ spec:
|
||||
# Job groups definitions
|
||||
groups:
|
||||
- name: "first-group"
|
||||
# Group will run in parallel with other defined groups
|
||||
parallel: true
|
||||
# Group specific parameters
|
||||
params:
|
||||
env:
|
||||
- name: "FEE"
|
||||
@@ -63,6 +60,11 @@ spec:
|
||||
args:
|
||||
- "sleep"
|
||||
- "10"
|
||||
- name: "second-half-job"
|
||||
image: "busybox"
|
||||
args:
|
||||
- "sleep"
|
||||
- "10"
|
||||
|
||||
- name: "second-group"
|
||||
parallel: true
|
||||
@@ -79,9 +81,44 @@ spec:
|
||||
args:
|
||||
- "sleep"
|
||||
- "10"
|
||||
parallel: false
|
||||
|
||||
- name: "third-group"
|
||||
parallel: false
|
||||
jobs:
|
||||
- name: "fifth-job"
|
||||
image: "busybox"
|
||||
args:
|
||||
- "echo"
|
||||
- "Hello world!"
|
||||
parallel: true
|
||||
```
|
||||
|
||||
### How does it look in practice?
|
||||
|
||||
```yaml
|
||||
managedjob-sample
|
||||
├── first-group
|
||||
│ ├── first-job
|
||||
│ ├── second-job
|
||||
│ │ └── Depends on: managedjob-sample-first-group-first-job
|
||||
│ └── second-half-job
|
||||
│ ├── Depends on: managedjob-sample-first-group-first-job
|
||||
│ └── Depends on: managedjob-sample-first-group-second-job
|
||||
├── second-group
|
||||
│ ├── third-job
|
||||
│ └── fourth-job
|
||||
│ └── Depends on: managedjob-sample-second-group-third-job
|
||||
└── third-group
|
||||
├── fifth-job
|
||||
├── Depends on group: first-group
|
||||
└── Depends on group: second-group
|
||||
```
|
||||
|
||||
If dependency exists on the group level - the group will not be executed until all of remaining groups have finished successfuly.
|
||||
If dependency exists on the job level - the job will not be executed until all of remaining jobs have finished successfuly.
|
||||
Remember that **ORDER matters**.
|
||||
|
||||
### Things to remember
|
||||
|
||||
Parameters **params** are always merged downwards to DRY your definitions.
|
||||
@@ -108,6 +145,65 @@ In this case - result for the first job will look like this:
|
||||
status: succeeded
|
||||
```
|
||||
|
||||
### Available params
|
||||
|
||||
There's quite a lot of of flexibility with parameters. On every level where params are allowed, you can define:
|
||||
|
||||
```yaml
|
||||
params:
|
||||
fromEnv:
|
||||
- configMapRef:
|
||||
name: "configmap-name"
|
||||
key: "key-name"
|
||||
env:
|
||||
- name: "FOO"
|
||||
value: "bar"
|
||||
volumes:
|
||||
- name: secrets-store-api
|
||||
csi:
|
||||
driver: secrets-store.csi.k8s.io
|
||||
readOnly: true
|
||||
volumeAttributes:
|
||||
secretProviderClass: api-secrets-provider
|
||||
volumeMount:
|
||||
- name: secrets-store-api
|
||||
mountPath: "/mnt/secrets-api"
|
||||
readOnly: true
|
||||
serviceAccount: "service-account-name"
|
||||
restartPolicy: "Never"
|
||||
imagePullSecrets:
|
||||
- "ghcr-token"
|
||||
imagePullPolicy:
|
||||
- "Always"
|
||||
labels:
|
||||
"label-name": "label-value"
|
||||
```
|
||||
|
||||
|
||||
### Kustomization and references
|
||||
|
||||
In case of any issues with `configmapGenerator` or `secretGenerator`, please add following to your `kustomization.yaml`:
|
||||
|
||||
```yaml
|
||||
configurations:
|
||||
- crd-name-reference.yaml
|
||||
```
|
||||
|
||||
Then you can create `crd-name-reference.yaml` file with following content:
|
||||
|
||||
```yaml
|
||||
---
|
||||
nameReference:
|
||||
- kind: 'ConfigMap'
|
||||
fieldSpecs:
|
||||
- kind: 'ManagedJob'
|
||||
path: 'spec/params/fromEnv[]/configMapRef/name'
|
||||
- kind: 'ManagedJob'
|
||||
path: 'spec/params/env[]/configMapRef/name'
|
||||
```
|
||||
|
||||
This will instruct kustomize to replace all references to configmaps with their names if they are managed by generators.
|
||||
|
||||
### Running on the cluster
|
||||
|
||||
#### Manual installation
|
||||
|
||||
@@ -24,10 +24,8 @@ import (
|
||||
type ManagedJobDependencies struct {
|
||||
// +kubebuilder:validation:Optional
|
||||
// +kubebuilder:default=""
|
||||
Group string `json:"group"`
|
||||
// +kubebuilder:validation:Optional
|
||||
// +kubebuilder:default=""
|
||||
Job string `json:"job"`
|
||||
Name string `json:"name"`
|
||||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
type ManagedJobDefinition struct {
|
||||
@@ -48,8 +46,9 @@ type ManagedJobDefinition struct {
|
||||
// +kubebuilder:validation:Optional
|
||||
// +kubebuilder:default=pending
|
||||
Status string `json:"status"`
|
||||
// +kubebuilder:validation:Optional
|
||||
// +optional
|
||||
Dependencies ManagedJobDependencies `json:"dependencies"`
|
||||
Dependencies []*ManagedJobDependencies `json:"dependencies"`
|
||||
// +optional
|
||||
CompiledParams ManagedJobParameters `json:"compiledParams"`
|
||||
}
|
||||
@@ -68,6 +67,9 @@ type ManagedJobGroup struct {
|
||||
// +kubebuilder:validation:Optional
|
||||
Params ManagedJobParameters `json:"params"`
|
||||
// +kubebuilder:validation:Optional
|
||||
// +optional
|
||||
Dependencies []*ManagedJobDependencies `json:"dependencies"`
|
||||
// +kubebuilder:validation:Optional
|
||||
// +kubebuilder:default=pending
|
||||
Status string `json:"status"`
|
||||
}
|
||||
@@ -92,6 +94,8 @@ type ManagedJobParameters struct {
|
||||
ImagePullPolicy string `json:"imagePullPolicy,omitempty"`
|
||||
// +kubebuilder:validation:Optional
|
||||
Labels map[string]string `json:"labels,omitempty"`
|
||||
// +kubebuilder:validation:Optional
|
||||
Annotations map[string]string `json:"annotations,omitempty"`
|
||||
}
|
||||
|
||||
// ManagedJobSpec defines the desired state of ManagedJob
|
||||
@@ -105,9 +109,6 @@ type ManagedJobSpec struct {
|
||||
Groups []*ManagedJobGroup `json:"groups"`
|
||||
// +kubebuilder:validation:Optional
|
||||
Params ManagedJobParameters `json:"params"`
|
||||
// +kubebuilder:validation:Optional
|
||||
// +kubebuilder:default=pending
|
||||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
// +kubebuilder:object:root=true
|
||||
|
||||
@@ -61,7 +61,17 @@ func (in *ManagedJobDefinition) DeepCopyInto(out *ManagedJobDefinition) {
|
||||
copy(*out, *in)
|
||||
}
|
||||
in.Params.DeepCopyInto(&out.Params)
|
||||
out.Dependencies = in.Dependencies
|
||||
if in.Dependencies != nil {
|
||||
in, out := &in.Dependencies, &out.Dependencies
|
||||
*out = make([]*ManagedJobDependencies, len(*in))
|
||||
for i := range *in {
|
||||
if (*in)[i] != nil {
|
||||
in, out := &(*in)[i], &(*out)[i]
|
||||
*out = new(ManagedJobDependencies)
|
||||
**out = **in
|
||||
}
|
||||
}
|
||||
}
|
||||
in.CompiledParams.DeepCopyInto(&out.CompiledParams)
|
||||
}
|
||||
|
||||
@@ -105,6 +115,17 @@ func (in *ManagedJobGroup) DeepCopyInto(out *ManagedJobGroup) {
|
||||
}
|
||||
}
|
||||
in.Params.DeepCopyInto(&out.Params)
|
||||
if in.Dependencies != nil {
|
||||
in, out := &in.Dependencies, &out.Dependencies
|
||||
*out = make([]*ManagedJobDependencies, len(*in))
|
||||
for i := range *in {
|
||||
if (*in)[i] != nil {
|
||||
in, out := &(*in)[i], &(*out)[i]
|
||||
*out = new(ManagedJobDependencies)
|
||||
**out = **in
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ManagedJobGroup.
|
||||
@@ -192,6 +213,13 @@ func (in *ManagedJobParameters) DeepCopyInto(out *ManagedJobParameters) {
|
||||
(*out)[key] = val
|
||||
}
|
||||
}
|
||||
if in.Annotations != nil {
|
||||
in, out := &in.Annotations, &out.Annotations
|
||||
*out = make(map[string]string, len(*in))
|
||||
for key, val := range *in {
|
||||
(*out)[key] = val
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ManagedJobParameters.
|
||||
|
||||
@@ -13,12 +13,12 @@ type: application
|
||||
# This is the chart version. This version number should be incremented each time you make changes
|
||||
# to the chart and its templates, including the app version.
|
||||
# Versions are expected to follow Semantic Versioning (https://semver.org/)
|
||||
version: 0.0.2
|
||||
version: 0.0.21
|
||||
# This is the version number of the application being deployed. This version number should be
|
||||
# incremented each time you make changes to the application. Versions are not expected to
|
||||
# follow Semantic Versioning. They should reflect the version the application is using.
|
||||
# It is recommended to use it with quotes.
|
||||
appVersion: "0.0.2"
|
||||
appVersion: "0.0.21"
|
||||
keywords:
|
||||
- operator
|
||||
- jobs
|
||||
|
||||
@@ -38,6 +38,18 @@ spec:
|
||||
groups:
|
||||
items:
|
||||
properties:
|
||||
dependencies:
|
||||
items:
|
||||
properties:
|
||||
name:
|
||||
default: ""
|
||||
type: string
|
||||
status:
|
||||
type: string
|
||||
required:
|
||||
- status
|
||||
type: object
|
||||
type: array
|
||||
jobs:
|
||||
items:
|
||||
properties:
|
||||
@@ -47,6 +59,10 @@ spec:
|
||||
type: array
|
||||
compiledParams:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable
|
||||
@@ -2100,14 +2116,17 @@ spec:
|
||||
type: array
|
||||
type: object
|
||||
dependencies:
|
||||
properties:
|
||||
group:
|
||||
default: ""
|
||||
type: string
|
||||
job:
|
||||
default: ""
|
||||
type: string
|
||||
type: object
|
||||
items:
|
||||
properties:
|
||||
name:
|
||||
default: ""
|
||||
type: string
|
||||
status:
|
||||
type: string
|
||||
required:
|
||||
- status
|
||||
type: object
|
||||
type: array
|
||||
image:
|
||||
minLength: 5
|
||||
type: string
|
||||
@@ -2120,6 +2139,10 @@ spec:
|
||||
type: boolean
|
||||
params:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable
|
||||
@@ -4190,6 +4213,10 @@ spec:
|
||||
type: boolean
|
||||
params:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable present
|
||||
@@ -6116,6 +6143,10 @@ spec:
|
||||
type: array
|
||||
params:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable present
|
||||
@@ -7912,9 +7943,6 @@ spec:
|
||||
default: 1
|
||||
minimum: 1
|
||||
type: integer
|
||||
status:
|
||||
default: pending
|
||||
type: string
|
||||
required:
|
||||
- groups
|
||||
- retries
|
||||
|
||||
@@ -44,6 +44,15 @@ rules:
|
||||
- patch
|
||||
- update
|
||||
- watch
|
||||
# Add access to events
|
||||
- apiGroups:
|
||||
- ""
|
||||
resources:
|
||||
- events
|
||||
verbs:
|
||||
- create
|
||||
- patch
|
||||
- update
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: ClusterRoleBinding
|
||||
|
||||
@@ -12,8 +12,8 @@ controllerManager:
|
||||
memory: 64Mi
|
||||
manager:
|
||||
image:
|
||||
repository: ghcr.io/lukaszraczylo/jobsmanager
|
||||
tag: 0.0.2
|
||||
repository: ghcr.io/lukaszraczylo/jobs-manager-operator
|
||||
tag: {{ .Chart.AppVersion }}
|
||||
resources:
|
||||
limits:
|
||||
cpu: 500m
|
||||
|
||||
@@ -38,6 +38,18 @@ spec:
|
||||
groups:
|
||||
items:
|
||||
properties:
|
||||
dependencies:
|
||||
items:
|
||||
properties:
|
||||
name:
|
||||
default: ""
|
||||
type: string
|
||||
status:
|
||||
type: string
|
||||
required:
|
||||
- status
|
||||
type: object
|
||||
type: array
|
||||
jobs:
|
||||
items:
|
||||
properties:
|
||||
@@ -47,6 +59,10 @@ spec:
|
||||
type: array
|
||||
compiledParams:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable
|
||||
@@ -2170,14 +2186,17 @@ spec:
|
||||
type: array
|
||||
type: object
|
||||
dependencies:
|
||||
properties:
|
||||
group:
|
||||
default: ""
|
||||
type: string
|
||||
job:
|
||||
default: ""
|
||||
type: string
|
||||
type: object
|
||||
items:
|
||||
properties:
|
||||
name:
|
||||
default: ""
|
||||
type: string
|
||||
status:
|
||||
type: string
|
||||
required:
|
||||
- status
|
||||
type: object
|
||||
type: array
|
||||
image:
|
||||
minLength: 5
|
||||
type: string
|
||||
@@ -2190,6 +2209,10 @@ spec:
|
||||
type: boolean
|
||||
params:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable
|
||||
@@ -4330,6 +4353,10 @@ spec:
|
||||
type: boolean
|
||||
params:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable
|
||||
@@ -6302,6 +6329,10 @@ spec:
|
||||
type: array
|
||||
params:
|
||||
properties:
|
||||
annotations:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
env:
|
||||
items:
|
||||
description: EnvVar represents an environment variable present
|
||||
@@ -8135,9 +8166,6 @@ spec:
|
||||
default: 1
|
||||
minimum: 1
|
||||
type: integer
|
||||
status:
|
||||
default: pending
|
||||
type: string
|
||||
required:
|
||||
- groups
|
||||
- retries
|
||||
|
||||
@@ -41,6 +41,11 @@ spec:
|
||||
args:
|
||||
- "sleep"
|
||||
- "10"
|
||||
- name: "second-half-job"
|
||||
image: "busybox"
|
||||
args:
|
||||
- "sleep"
|
||||
- "10"
|
||||
|
||||
- name: "second-group"
|
||||
parallel: true
|
||||
@@ -57,4 +62,14 @@ spec:
|
||||
args:
|
||||
- "sleep"
|
||||
- "10"
|
||||
parallel: false
|
||||
|
||||
- name: "third-group"
|
||||
parallel: false
|
||||
jobs:
|
||||
- name: "fifth-job"
|
||||
image: "busybox"
|
||||
args:
|
||||
- "echo"
|
||||
- "Hello world!"
|
||||
parallel: true
|
||||
@@ -1,71 +0,0 @@
|
||||
package controllers
|
||||
|
||||
import (
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
)
|
||||
|
||||
func (cp *connPackage) checkGroupsStatus() {
|
||||
groupsTotal := len(cp.mj.Spec.Groups)
|
||||
totalJobs := 0
|
||||
completedJobs := 0
|
||||
didAnyJobAbort := false
|
||||
changePresent := false
|
||||
groupsCompleted := 0
|
||||
|
||||
// Check if all groups have completed and set ManagedJob status to "succeeded" if so.
|
||||
if cp.mj.Spec.Status != ExecutionStatusSucceeded && groupsTotal > 0 {
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
groupJobsTotal := len(group.Jobs)
|
||||
groupJobsCompleted := 0
|
||||
|
||||
for _, job := range group.Jobs {
|
||||
if job.Status == ExecutionStatusSucceeded {
|
||||
groupJobsCompleted++
|
||||
completedJobs++
|
||||
}
|
||||
if job.Status == ExecutionStatusFailed || job.Status == ExecutionStatusAborted {
|
||||
didAnyJobAbort = true
|
||||
}
|
||||
totalJobs++
|
||||
}
|
||||
|
||||
if groupJobsTotal == groupJobsCompleted {
|
||||
// All the jobs in the group are completed.
|
||||
if group.Status != ExecutionStatusSucceeded {
|
||||
group.Status = ExecutionStatusSucceeded
|
||||
changePresent = true
|
||||
}
|
||||
groupsCompleted++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if groupsTotal == groupsCompleted && cp.mj.Spec.Status != ExecutionStatusSucceeded && cp.mj.Status != ExecutionStatusSucceeded {
|
||||
cp.mj.Spec.Status = ExecutionStatusSucceeded
|
||||
changePresent = true
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Completed", "All jobs completed")
|
||||
}
|
||||
|
||||
// Update status if any job aborted.
|
||||
if didAnyJobAbort && cp.mj.Spec.Status != ExecutionStatusFailed {
|
||||
cp.mj.Spec.Status = ExecutionStatusFailed
|
||||
changePresent = true
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Aborted", "One of the jobs aborted")
|
||||
}
|
||||
|
||||
// Update status to "running" if not already set.
|
||||
// if cp.mj.Spec.Status != ExecutionStatusRunning && cp.mj.Spec.Status != ExecutionStatusSucceeded {
|
||||
// cp.mj.Spec.Status = ExecutionStatusRunning
|
||||
// cp.mj.Status = ExecutionStatusRunning
|
||||
// changePresent = true
|
||||
// }
|
||||
|
||||
// Check if the ManagedJob status has changed.
|
||||
statusChanged := cp.mj.Spec.Status != cp.mj.Status
|
||||
|
||||
// Update status and send event if it has changed.
|
||||
if statusChanged || changePresent {
|
||||
cp.updateCRDStatusDirectly()
|
||||
// cp.r.Client.Status().Update(cp.ctx, cp.mj)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
package controllers
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/lukaszraczylo/pandati"
|
||||
jobsmanagerv1beta1 "raczylo.com/jobs-manager-operator/api/v1beta1"
|
||||
)
|
||||
|
||||
const (
|
||||
newLine = "\n"
|
||||
emptySpace = " "
|
||||
middleItem = "├── "
|
||||
continueItem = "│ "
|
||||
lastItem = "└── "
|
||||
)
|
||||
|
||||
func New(text string) Tree {
|
||||
return &tree{
|
||||
text: text,
|
||||
items: []Tree{},
|
||||
}
|
||||
}
|
||||
|
||||
// Add adds a node to the tree
|
||||
func (t *tree) Add(text string) Tree {
|
||||
n := New(text)
|
||||
t.items = append(t.items, n)
|
||||
return n
|
||||
}
|
||||
|
||||
// AddTree adds a tree as an item
|
||||
func (t *tree) AddTree(tree Tree) {
|
||||
t.items = append(t.items, tree)
|
||||
}
|
||||
|
||||
// Text returns the node's value
|
||||
func (t *tree) Text() string {
|
||||
return t.text
|
||||
}
|
||||
|
||||
// Items returns all items in the tree
|
||||
func (t *tree) Items() []Tree {
|
||||
return t.items
|
||||
}
|
||||
|
||||
func (t *tree) Print() string {
|
||||
p := &printer{}
|
||||
return p.Print(t)
|
||||
}
|
||||
|
||||
func (p *printer) Print(t Tree) string {
|
||||
return t.Text() + newLine + p.printItems(t.Items(), []bool{})
|
||||
}
|
||||
|
||||
func (p *printer) printText(text string, spaces []bool, last bool) string {
|
||||
var result string
|
||||
for _, space := range spaces {
|
||||
if space {
|
||||
result += emptySpace
|
||||
} else {
|
||||
result += continueItem
|
||||
}
|
||||
}
|
||||
|
||||
indicator := middleItem
|
||||
if last {
|
||||
indicator = lastItem
|
||||
}
|
||||
|
||||
var out string
|
||||
lines := strings.Split(text, "\n")
|
||||
for i := range lines {
|
||||
text := lines[i]
|
||||
if i == 0 {
|
||||
out += result + indicator + text + newLine
|
||||
continue
|
||||
}
|
||||
if last {
|
||||
indicator = emptySpace
|
||||
} else {
|
||||
indicator = continueItem
|
||||
}
|
||||
out += result + indicator + text + newLine
|
||||
}
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
func (p *printer) printItems(t []Tree, spaces []bool) string {
|
||||
var result string
|
||||
for i, f := range t {
|
||||
last := i == len(t)-1
|
||||
result += p.printText(f.Text(), spaces, last)
|
||||
if len(f.Items()) > 0 {
|
||||
spacesChild := append(spaces, last)
|
||||
result += p.printItems(f.Items(), spacesChild)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func (cp *connPackage) checkIfPresentInDependencies(currentDependencies []*jobsmanagerv1beta1.ManagedJobDependencies, dependencyName string) bool {
|
||||
for _, dependency := range currentDependencies {
|
||||
if dependency.Name == dependencyName {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (cp *connPackage) generateDependencyTree() {
|
||||
// First pass - initialize the tree and get all the gathered jobs
|
||||
originalMainJobDefinition := cp.mj.DeepCopy()
|
||||
|
||||
mainTree := New(cp.mj.Name)
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
groupTree := mainTree.Add(group.Name)
|
||||
for _, job := range group.Jobs {
|
||||
jobTree := groupTree.Add(job.Name)
|
||||
job.CompiledParams = cp.compileParameters(cp.mj.Spec.Params, group.Params, job.Params)
|
||||
if job.Parallel {
|
||||
continue
|
||||
} else {
|
||||
// get the groupTree items before this job and add them as dependencies
|
||||
for _, jobTreePrevious := range groupTree.Items() {
|
||||
if jobTreePrevious.Text() == job.Name {
|
||||
break
|
||||
}
|
||||
generatedJobName := jobNameGenerator(cp.mj.Name, group.Name, jobTreePrevious.Text())
|
||||
jobTree.Add("Depends on: " + generatedJobName)
|
||||
if !cp.checkIfPresentInDependencies(job.Dependencies, generatedJobName) {
|
||||
job.Dependencies = append(job.Dependencies, &jobsmanagerv1beta1.ManagedJobDependencies{Name: generatedJobName, Status: ExecutionStatusPending})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if group.Parallel {
|
||||
continue
|
||||
} else {
|
||||
// get the mainTree items before this group and add them as dependencies
|
||||
for _, groupTreePrevious := range mainTree.Items() {
|
||||
if groupTreePrevious.Text() == group.Name {
|
||||
break
|
||||
}
|
||||
groupTree.Add("Depends on group: " + groupTreePrevious.Text())
|
||||
if !cp.checkIfPresentInDependencies(group.Dependencies, groupTreePrevious.Text()) {
|
||||
group.Dependencies = append(group.Dependencies, &jobsmanagerv1beta1.ManagedJobDependencies{Name: groupTreePrevious.Text(), Status: ExecutionStatusPending})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_, theSame, _ := pandati.CompareStructsReplaced(originalMainJobDefinition, cp.mj)
|
||||
if !theSame {
|
||||
cp.updateCRDStatusDirectly()
|
||||
}
|
||||
// fmt.Print(mainTree.Print())
|
||||
// fmt.Printf("Dependency tree: %# v", pretty.Formatter(mainTree))
|
||||
}
|
||||
+156
-86
@@ -1,6 +1,8 @@
|
||||
package controllers
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/lukaszraczylo/pandati"
|
||||
kbatch "k8s.io/api/batch/v1"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
@@ -22,6 +24,7 @@ type compiledParams struct {
|
||||
ImagePullSecrets []corev1.LocalObjectReference
|
||||
ImagePullPolicy string
|
||||
Labels map[string]string
|
||||
Annotations map[string]string
|
||||
}
|
||||
|
||||
func (cp *connPackage) compileParameters(params ...jobsmanagerv1beta1.ManagedJobParameters) jobsmanagerv1beta1.ManagedJobParameters {
|
||||
@@ -34,9 +37,9 @@ func (cp *connPackage) compileParameters(params ...jobsmanagerv1beta1.ManagedJob
|
||||
if params.Env != nil {
|
||||
cparams.Env = append(cparams.Env, params.Env...)
|
||||
}
|
||||
// if params.Volumes != nil {
|
||||
// cparams.Volumes = append(cparams.Volumes, params.Volumes...)
|
||||
// }
|
||||
if params.Volumes != nil {
|
||||
cparams.Volumes = append(cparams.Volumes, params.Volumes...)
|
||||
}
|
||||
if params.VolumeMounts != nil {
|
||||
cparams.VolumeMounts = append(cparams.VolumeMounts, params.VolumeMounts...)
|
||||
}
|
||||
@@ -54,140 +57,176 @@ func (cp *connPackage) compileParameters(params ...jobsmanagerv1beta1.ManagedJob
|
||||
cparams.Labels[k] = v
|
||||
}
|
||||
}
|
||||
if params.Annotations != nil {
|
||||
for k, v := range params.Annotations {
|
||||
cparams.Annotations[k] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return cparams
|
||||
}
|
||||
|
||||
type previousJobsAndGroups struct {
|
||||
GroupID string
|
||||
JobID string
|
||||
}
|
||||
|
||||
func (cp *connPackage) buildJobsDependencyTree() error {
|
||||
changePending := false
|
||||
for i, group := range cp.mj.Spec.Groups {
|
||||
var group_previous string
|
||||
if group.Parallel {
|
||||
group_previous = group.Name
|
||||
} else {
|
||||
if i > 0 {
|
||||
group_previous = cp.mj.Spec.Groups[i-1].Name
|
||||
} else {
|
||||
group_previous = group.Name
|
||||
}
|
||||
}
|
||||
for j, job := range group.Jobs {
|
||||
if job.Dependencies.Group == "" || job.Dependencies.Job == "" {
|
||||
changePending = true
|
||||
}
|
||||
job.Dependencies.Group = group_previous
|
||||
if job.Parallel {
|
||||
job.Dependencies.Job = job.Name
|
||||
} else {
|
||||
if j > 0 {
|
||||
job.Dependencies.Job = group.Jobs[j-1].Name
|
||||
} else {
|
||||
job.Dependencies.Job = job.Name
|
||||
func (cp *connPackage) updateDependentJobs(completedJob string, jobStatus string) {
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
for _, job := range group.Jobs {
|
||||
for _, dependency := range job.Dependencies {
|
||||
if dependency.Name == completedJob && dependency.Status != jobStatus {
|
||||
dependency.Status = jobStatus
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if changePending {
|
||||
cp.updateCRDStatusDirectly()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cp *connPackage) checkJobStatus() {
|
||||
func (cp *connPackage) updateDependentGroups(completedGroup string, jobStatus string) {
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
for _, dependency := range group.Dependencies {
|
||||
if dependency.Name == completedGroup && dependency.Status != jobStatus {
|
||||
dependency.Status = jobStatus
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (cp *connPackage) checkRunningJobsStatus() {
|
||||
var childJobs kbatch.JobList
|
||||
labelSelector := labels.SelectorFromSet(labels.Set{
|
||||
"jobmanager.raczylo.com/workflow-name": cp.mj.Name,
|
||||
})
|
||||
listOptions := &client.ListOptions{LabelSelector: labelSelector, Namespace: cp.mj.Namespace}
|
||||
|
||||
err := cp.r.Client.List(cp.ctx, &childJobs, listOptions)
|
||||
if err != nil {
|
||||
// log.Log.Error(err, "unable to list child jobs")
|
||||
log.Log.Info("Unable to list child jobs", "error", err.Error())
|
||||
return
|
||||
}
|
||||
changePresent := false
|
||||
|
||||
for _, childJob := range childJobs.Items {
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
for _, job := range group.Jobs {
|
||||
generatedJobName := jobNameGenerator(cp.mj.Name, group.Name, job.Name)
|
||||
if childJob.Name == generatedJobName {
|
||||
if childJob.Status.Succeeded > 0 && job.Status != ExecutionStatusSucceeded {
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Completed", "Job %s completed [prev: %s]", childJob.Name, job.Status)
|
||||
job.Status = ExecutionStatusSucceeded
|
||||
changePresent = true
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Completed", "Job %s completed", childJob.Name)
|
||||
} else if childJob.Status.Failed > 0 && job.Status != ExecutionStatusFailed {
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeWarning, "Failed", "Job %s failed [prev: %s]", childJob.Name, job.Status)
|
||||
job.Status = ExecutionStatusFailed
|
||||
changePresent = true
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Failed", "Job %s failed", childJob.Name)
|
||||
} else if childJob.Status.Active > 0 && job.Status != ExecutionStatusRunning {
|
||||
changePresent = true
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Running", "Job %s running [prev: %s]", childJob.Name, job.Status)
|
||||
job.Status = ExecutionStatusRunning
|
||||
}
|
||||
cp.updateDependentJobs(generatedJobName, job.Status)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
if changePresent {
|
||||
cp.updateCRDStatusDirectly()
|
||||
}
|
||||
}
|
||||
|
||||
func (cp *connPackage) runPendingJobs() {
|
||||
copyMJ := cp.mj.DeepCopy()
|
||||
changePresent := false
|
||||
// originalMainJobDefinition := cp.mj.DeepCopy()
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
run_group := false
|
||||
|
||||
groupJobsCompleted := 0
|
||||
for _, job := range group.Jobs {
|
||||
if job.Status == ExecutionStatusPending {
|
||||
if job.Dependencies.Group == group.Name && job.Dependencies.Job == job.Name {
|
||||
job.CompiledParams = cp.compileParameters(job.Params, group.Params, cp.mj.Spec.Params)
|
||||
err := cp.executeJob(job, group)
|
||||
if err != nil {
|
||||
// log.Log.Info("Unable to execute job", "group", group.Name, "job", job.Name, "error", err)
|
||||
continue
|
||||
if job.Status == ExecutionStatusSucceeded {
|
||||
groupJobsCompleted++
|
||||
}
|
||||
}
|
||||
if groupJobsCompleted == len(group.Jobs) {
|
||||
group.Status = ExecutionStatusSucceeded
|
||||
cp.updateDependentGroups(group.Name, group.Status)
|
||||
continue
|
||||
}
|
||||
|
||||
approvedStatuses := []string{ExecutionStatusSucceeded, ExecutionStatusFailed, ExecutionStatusAborted}
|
||||
if pandati.ExistsInSlice(approvedStatuses, group.Status) {
|
||||
cp.updateDependentGroups(group.Name, group.Status)
|
||||
}
|
||||
|
||||
approvedStatuses = []string{ExecutionStatusPending, ExecutionStatusRunning}
|
||||
if pandati.ExistsInSlice(approvedStatuses, group.Status) {
|
||||
if len(group.Dependencies) > 0 {
|
||||
groupsCompleted := 0
|
||||
for _, group_dependency := range group.Dependencies {
|
||||
if group_dependency.Status == ExecutionStatusSucceeded {
|
||||
groupsCompleted++
|
||||
}
|
||||
job.Status = ExecutionStatusRunning
|
||||
changePresent = true
|
||||
continue
|
||||
} else {
|
||||
for _, group2 := range copyMJ.Spec.Groups {
|
||||
for _, job2 := range group2.Jobs {
|
||||
if job2.Name == job.Dependencies.Job && group2.Name == job.Dependencies.Group {
|
||||
switch job2.Status {
|
||||
case ExecutionStatusSucceeded:
|
||||
job.CompiledParams = cp.compileParameters(job.Params, group.Params, cp.mj.Spec.Params)
|
||||
err := cp.executeJob(job, group)
|
||||
if err != nil {
|
||||
log.Log.Info("Unable to execute job", "group", group.Name, "job", job.Name, "error", err)
|
||||
continue
|
||||
}
|
||||
job.Status = ExecutionStatusRunning
|
||||
changePresent = true
|
||||
case ExecutionStatusFailed:
|
||||
if group_dependency.Status == ExecutionStatusFailed {
|
||||
group.Status = ExecutionStatusAborted
|
||||
cp.updateDependentGroups(group.Name, ExecutionStatusFailed)
|
||||
}
|
||||
}
|
||||
if groupsCompleted == len(group.Dependencies) {
|
||||
run_group = true
|
||||
}
|
||||
} else {
|
||||
run_group = true
|
||||
}
|
||||
|
||||
if !run_group {
|
||||
// fmt.Println("Group "+group.Name+" is not running as dependencies were not met", group.Dependencies)
|
||||
continue // not running the group as dependencies were not met
|
||||
} else {
|
||||
group.Status = ExecutionStatusRunning
|
||||
cp.updateDependentGroups(group.Name, ExecutionStatusRunning)
|
||||
|
||||
for _, job := range group.Jobs {
|
||||
run_job := false
|
||||
if job.Status == ExecutionStatusPending {
|
||||
if len(job.Dependencies) > 0 {
|
||||
jobsCompleted := 0
|
||||
for _, job_dependency := range job.Dependencies {
|
||||
if job_dependency.Status == ExecutionStatusSucceeded {
|
||||
jobsCompleted++
|
||||
}
|
||||
if job_dependency.Status == ExecutionStatusFailed {
|
||||
job.Status = ExecutionStatusAborted
|
||||
changePresent = true
|
||||
cp.updateDependentJobs(job.Name, ExecutionStatusFailed)
|
||||
}
|
||||
}
|
||||
if jobsCompleted == len(job.Dependencies) {
|
||||
run_job = true
|
||||
}
|
||||
} else {
|
||||
run_job = true
|
||||
}
|
||||
}
|
||||
|
||||
if !run_job {
|
||||
continue // job is not ready as dependencies were not met
|
||||
} else {
|
||||
approvedStatuses = []string{ExecutionStatusRunning, ExecutionStatusFailed, ExecutionStatusAborted}
|
||||
if !pandati.ExistsInSlice(approvedStatuses, job.Status) {
|
||||
err := cp.executeJob(job, group)
|
||||
if err != nil {
|
||||
log.Log.Info("Unable to execute job", "error", err.Error())
|
||||
if !strings.Contains(err.Error(), "exists") {
|
||||
job.Status = ExecutionStatusFailed
|
||||
group.Status = ExecutionStatusFailed
|
||||
cp.updateDependentJobs(job.Name, ExecutionStatusFailed)
|
||||
cp.updateDependentGroups(group.Name, ExecutionStatusFailed)
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeWarning, "Failed", "Job %s from group %s failed", job.Name, group.Name)
|
||||
}
|
||||
return
|
||||
}
|
||||
job.Status = ExecutionStatusRunning
|
||||
cp.updateDependentJobs(job.Name, ExecutionStatusRunning)
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Running", "Job %s from group %s running", job.Name, group.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fmt.Println("Running group: ", group.Name, " with status: ", group.Status, " accepted: ", run_group)
|
||||
}
|
||||
}
|
||||
if changePresent {
|
||||
cp.updateCRDStatusDirectly()
|
||||
}
|
||||
}
|
||||
|
||||
func (cp *connPackage) executeJob(j *jobsmanagerv1beta1.ManagedJobDefinition, g *jobsmanagerv1beta1.ManagedJobGroup) (err error) {
|
||||
generatedJobName := jobNameGenerator(cp.mj.Name, g.Name, j.Name)
|
||||
|
||||
convertRetries := func(retries int) *int32 {
|
||||
if retries == 0 {
|
||||
return nil
|
||||
@@ -209,6 +248,12 @@ func (cp *connPackage) executeJob(j *jobsmanagerv1beta1.ManagedJobDefinition, g
|
||||
labels[k] = v
|
||||
}
|
||||
|
||||
annotations := map[string]string{}
|
||||
|
||||
for k, v := range j.CompiledParams.Annotations {
|
||||
annotations[k] = v
|
||||
}
|
||||
|
||||
job_handler := kbatch.Job{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: generatedJobName,
|
||||
@@ -217,9 +262,10 @@ func (cp *connPackage) executeJob(j *jobsmanagerv1beta1.ManagedJobDefinition, g
|
||||
Spec: kbatch.JobSpec{
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: generatedJobName,
|
||||
Namespace: cp.mj.Namespace,
|
||||
Labels: labels,
|
||||
Name: generatedJobName,
|
||||
Namespace: cp.mj.Namespace,
|
||||
Labels: labels,
|
||||
Annotations: annotations,
|
||||
},
|
||||
Spec: corev1.PodSpec{
|
||||
Volumes: j.CompiledParams.Volumes,
|
||||
@@ -252,10 +298,34 @@ func (cp *connPackage) executeJob(j *jobsmanagerv1beta1.ManagedJobDefinition, g
|
||||
|
||||
err = cp.r.Client.Create(cp.ctx, &job_handler)
|
||||
if err != nil || pandati.IsZero(job_handler) {
|
||||
log.Log.Error(err, "Unable to create job", "job", job_handler.Name)
|
||||
return err
|
||||
}
|
||||
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Created", "Created job %s", job_handler.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cp *connPackage) checkOverallStatus() {
|
||||
groupsCompleted := 0
|
||||
negativeStatuses := []string{ExecutionStatusFailed, ExecutionStatusAborted}
|
||||
for _, group := range cp.mj.Spec.Groups {
|
||||
if group.Status == ExecutionStatusSucceeded {
|
||||
groupsCompleted++
|
||||
} else if pandati.ExistsInSlice(negativeStatuses, group.Status) {
|
||||
cp.mj.Status = ExecutionStatusFailed
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeWarning, "Failure", "Run failed in group %s", group.Name)
|
||||
} else {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
if groupsCompleted == len(cp.mj.Spec.Groups) {
|
||||
if cp.mj.Status != ExecutionStatusSucceeded {
|
||||
cp.r.Recorder.Eventf(cp.mj, corev1.EventTypeNormal, "Success", "Run completed successfuly")
|
||||
}
|
||||
cp.mj.Status = ExecutionStatusSucceeded
|
||||
} else {
|
||||
cp.mj.Status = ExecutionStatusRunning
|
||||
}
|
||||
cp.r.Status().Update(cp.ctx, cp.mj)
|
||||
}
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
package controllers
|
||||
|
||||
// +kubebuilder:validation:Enum=Allow;Forbid;Replace
|
||||
type ExecutionStatus string
|
||||
|
||||
const (
|
||||
ExecutionStatusPending string = "pending"
|
||||
ExecutionStatusRunning string = "running"
|
||||
@@ -15,3 +13,29 @@ const (
|
||||
var (
|
||||
jobOwnerKey = ".metadata.controller"
|
||||
)
|
||||
|
||||
type (
|
||||
ExecutionStatus string
|
||||
|
||||
tree struct {
|
||||
text string
|
||||
items []Tree
|
||||
}
|
||||
|
||||
// Tree is tree interface
|
||||
Tree interface {
|
||||
Add(text string) Tree
|
||||
AddTree(tree Tree)
|
||||
Items() []Tree
|
||||
Text() string
|
||||
Print() string
|
||||
}
|
||||
|
||||
printer struct {
|
||||
}
|
||||
|
||||
// Printer is printer interface
|
||||
Printer interface {
|
||||
Print(Tree) string
|
||||
}
|
||||
)
|
||||
|
||||
+6
-69
@@ -2,19 +2,13 @@ package controllers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/kr/pretty"
|
||||
"github.com/lukaszraczylo/pandati"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"raczylo.com/jobs-manager-operator/api/v1beta1"
|
||||
jobsmanagerv1beta1 "raczylo.com/jobs-manager-operator/api/v1beta1"
|
||||
ctrl "sigs.k8s.io/controller-runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
"sigs.k8s.io/controller-runtime/pkg/log"
|
||||
)
|
||||
|
||||
@@ -30,11 +24,12 @@ type jobStatusUpdate struct {
|
||||
}
|
||||
|
||||
type connPackage struct {
|
||||
r *ManagedJobReconciler
|
||||
ctx context.Context
|
||||
req ctrl.Request
|
||||
mtx sync.Mutex
|
||||
mj *jobsmanagerv1beta1.ManagedJob
|
||||
r *ManagedJobReconciler
|
||||
ctx context.Context
|
||||
req ctrl.Request
|
||||
mtx sync.Mutex
|
||||
mj *jobsmanagerv1beta1.ManagedJob
|
||||
dependencyTree Tree
|
||||
}
|
||||
|
||||
func (cp *connPackage) getOwnerReference() (metav1.OwnerReference, error) {
|
||||
@@ -53,65 +48,8 @@ func (cp *connPackage) getOwnerReference() (metav1.OwnerReference, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
type patchStringValue struct {
|
||||
Op string `json:"op"`
|
||||
Path string `json:"path"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
|
||||
func (cp *connPackage) getLatestMainJobAndPatch(proposedChanges *jobsmanagerv1beta1.ManagedJob) (bool, error) {
|
||||
mj := jobsmanagerv1beta1.ManagedJob{}
|
||||
err := cp.r.Client.Get(cp.ctx, cp.req.NamespacedName, &mj)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
diff, diffIdentical, _ := pandati.CompareStructsReplaced(mj, proposedChanges)
|
||||
if !diffIdentical {
|
||||
if err != nil {
|
||||
log.Log.Error(err, "Unable to marshal ManagedJob")
|
||||
}
|
||||
var payloadBytesArr []patchStringValue
|
||||
for _, d := range diff {
|
||||
if strings.HasPrefix(d.Key, "/spec") {
|
||||
operation := "replace"
|
||||
if d.OldValue == "" || d.OldValue == nil {
|
||||
operation = "add"
|
||||
}
|
||||
|
||||
payload := patchStringValue{
|
||||
Op: operation,
|
||||
Path: d.Key,
|
||||
Value: d.Value.(string),
|
||||
}
|
||||
payloadBytesArr = append(payloadBytesArr, payload)
|
||||
}
|
||||
}
|
||||
patchPayload, err := json.Marshal(payloadBytesArr)
|
||||
if err != nil {
|
||||
log.Log.Error(err, "Unable to marshal ManagedJob")
|
||||
}
|
||||
fmt.Printf("PatchPayload: %# v", pretty.Formatter(fmt.Sprintf("%s", patchPayload)))
|
||||
kubepath := client.RawPatch(types.JSONPatchType, patchPayload)
|
||||
err = cp.r.Client.Patch(cp.ctx, &mj, kubepath)
|
||||
// if err != nil {
|
||||
// log.Log.Error(err, "Unable to patch ManagedJob")
|
||||
// }
|
||||
return true, err
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
|
||||
func (cp *connPackage) updateCRDStatusDirectly() error {
|
||||
cp.mtx.Lock()
|
||||
// defer cp.mtx.Unlock()
|
||||
|
||||
// val, err := cp.getLatestMainJobAndPatch(cp.mj)
|
||||
// if err != nil {
|
||||
// log.Log.Error(err, "Unable to get latest ManagedJob")
|
||||
// return err
|
||||
// }
|
||||
|
||||
// if !val && err == nil {
|
||||
err := cp.r.Update(cp.ctx, cp.mj)
|
||||
if err != nil {
|
||||
// log.Log.Info("Error", err.Error(), "more", "Unable to update ManagedJob status directly")
|
||||
@@ -121,7 +59,6 @@ func (cp *connPackage) updateCRDStatusDirectly() error {
|
||||
if err != nil {
|
||||
log.Log.Error(err, "Unable to get updated ManagedJob")
|
||||
}
|
||||
// }
|
||||
cp.mtx.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ package controllers
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/lukaszraczylo/pandati"
|
||||
kbatch "k8s.io/api/batch/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/client-go/tools/record"
|
||||
@@ -44,9 +45,10 @@ func (r *ManagedJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
|
||||
_ = log.FromContext(ctx)
|
||||
|
||||
cp := &connPackage{
|
||||
r: r,
|
||||
ctx: ctx,
|
||||
req: req,
|
||||
r: r,
|
||||
ctx: ctx,
|
||||
req: req,
|
||||
dependencyTree: nil,
|
||||
}
|
||||
|
||||
var managedJob jobsmanagerv1beta1.ManagedJob
|
||||
@@ -55,13 +57,27 @@ func (r *ManagedJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
|
||||
}
|
||||
|
||||
cp.mj = &managedJob
|
||||
cp.buildJobsDependencyTree()
|
||||
cp.checkJobStatus()
|
||||
|
||||
originalMainJobDefinition := cp.mj.DeepCopy()
|
||||
cp.generateDependencyTree()
|
||||
_, theSame, _ := pandati.CompareStructsReplaced(originalMainJobDefinition, cp.mj)
|
||||
if !theSame {
|
||||
cp.updateCRDStatusDirectly()
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
originalMainJobDefinition = cp.mj.DeepCopy()
|
||||
|
||||
// TODO: Re-enable after testing
|
||||
cp.checkRunningJobsStatus()
|
||||
cp.runPendingJobs()
|
||||
cp.checkGroupsStatus()
|
||||
|
||||
_, theSame, _ = pandati.CompareStructsReplaced(originalMainJobDefinition, cp.mj)
|
||||
if !theSame {
|
||||
cp.updateCRDStatusDirectly()
|
||||
}
|
||||
|
||||
cp.checkOverallStatus()
|
||||
// fmt.Printf("Reconcile: %# v", pretty.Formatter(r.Updater))
|
||||
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -47,7 +47,7 @@ require (
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/prometheus/client_golang v1.14.0 // indirect
|
||||
github.com/prometheus/client_model v0.3.0 // indirect
|
||||
github.com/prometheus/common v0.39.0 // indirect
|
||||
github.com/prometheus/common v0.40.0 // indirect
|
||||
github.com/prometheus/procfs v0.9.0 // indirect
|
||||
github.com/rogpeppe/go-internal v1.9.0 // indirect
|
||||
github.com/rs/zerolog v1.29.0 // indirect
|
||||
@@ -75,7 +75,7 @@ require (
|
||||
k8s.io/component-base v0.26.1 // indirect
|
||||
k8s.io/klog/v2 v2.90.0 // indirect
|
||||
k8s.io/kube-openapi v0.0.0-20230217203603-ff9a8e8fa21d // indirect
|
||||
k8s.io/utils v0.0.0-20230209194617-a36077c30491 // indirect
|
||||
k8s.io/utils v0.0.0-20230220204549-a5ecb0141aa5 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
|
||||
sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
|
||||
sigs.k8s.io/yaml v1.3.0 // indirect
|
||||
|
||||
@@ -292,6 +292,8 @@ github.com/prometheus/common v0.37.0 h1:ccBbHCgIiT9uSoFY0vX8H3zsNR5eLt17/RQLUvn8
|
||||
github.com/prometheus/common v0.37.0/go.mod h1:phzohg0JFMnBEFGxTDbfu3QyL5GI8gTQJFhYO5B3mfA=
|
||||
github.com/prometheus/common v0.39.0 h1:oOyhkDq05hPZKItWVBkJ6g6AtGxi+fy7F4JvUV8uhsI=
|
||||
github.com/prometheus/common v0.39.0/go.mod h1:6XBZ7lYdLCbkAVhwRsWTZn+IN5AB9F/NXd5w0BbEX0Y=
|
||||
github.com/prometheus/common v0.40.0 h1:Afz7EVRqGg2Mqqf4JuF9vdvp1pi220m55Pi9T2JnO4Q=
|
||||
github.com/prometheus/common v0.40.0/go.mod h1:L65ZJPSmfn/UBWLQIHV7dBrKFidB/wPlF1y5TlSt9OE=
|
||||
github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
|
||||
github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
|
||||
github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
|
||||
@@ -725,6 +727,8 @@ k8s.io/utils v0.0.0-20221128185143-99ec85e7a448 h1:KTgPnR10d5zhztWptI952TNtt/4u5
|
||||
k8s.io/utils v0.0.0-20221128185143-99ec85e7a448/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
|
||||
k8s.io/utils v0.0.0-20230209194617-a36077c30491 h1:r0BAOLElQnnFhE/ApUsg3iHdVYYPBjNSSOMowRZxxsY=
|
||||
k8s.io/utils v0.0.0-20230209194617-a36077c30491/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
|
||||
k8s.io/utils v0.0.0-20230220204549-a5ecb0141aa5 h1:kmDqav+P+/5e1i9tFfHq1qcF3sOrDp+YEkVDAHu7Jwk=
|
||||
k8s.io/utils v0.0.0-20230220204549-a5ecb0141aa5/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
|
||||
rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
|
||||
rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
|
||||
rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
|
||||
|
||||
Reference in New Issue
Block a user