From 63cfac292e756d47641d1f632ba8fb90ba856f6a Mon Sep 17 00:00:00 2001 From: Jakub Senko Date: Fri, 26 May 2023 14:21:16 +0200 Subject: [PATCH] feat: support user-provided pod template (PodSpec) in the ApicurioRegistry CRD (preview feature) --- api/v1/apicurioregistry_types.go | 159 ++++++++- api/v1/zz_generated.deepcopy.go | 236 +++++++++++++ controllers/apicurioregistry_controller.go | 24 +- controllers/cf/cf_https.go | 2 +- controllers/cf/cf_image.go | 2 + controllers/cf/cf_pod_spec.go | 318 ++++++++++++++++++ controllers/common/util.go | 13 +- controllers/svc/factory/factory_kube.go | 79 ++--- .../condition_configuration_error.go | 4 +- 9 files changed, 765 insertions(+), 72 deletions(-) create mode 100644 controllers/cf/cf_pod_spec.go diff --git a/api/v1/apicurioregistry_types.go b/api/v1/apicurioregistry_types.go index 5312a235..4fa35fd2 100644 --- a/api/v1/apicurioregistry_types.go +++ b/api/v1/apicurioregistry_types.go @@ -17,8 +17,8 @@ limitations under the License. package v1 import ( - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + core "k8s.io/api/core/v1" + meta "k8s.io/apimachinery/pkg/apis/meta/v1" ) // ### Spec @@ -36,7 +36,7 @@ type ApicurioRegistrySpecConfiguration struct { UI ApicurioRegistrySpecConfigurationUI `json:"ui,omitempty"` LogLevel string `json:"logLevel,omitempty"` Security ApicurioRegistrySpecConfigurationSecurity `json:"security,omitempty"` - Env []corev1.EnvVar `json:"env,omitempty"` + Env []core.EnvVar `json:"env,omitempty"` } type ApicurioRegistrySpecConfigurationDataSource struct { @@ -100,18 +100,149 @@ type ApicurioRegistrySpecDeploymentMetadata struct { } type ApicurioRegistrySpecDeployment struct { - Replicas int32 `json:"replicas,omitempty"` - Host string `json:"host,omitempty"` - Affinity *corev1.Affinity `json:"affinity,omitempty"` - Tolerations []corev1.Toleration `json:"tolerations,omitempty"` + Replicas int32 `json:"replicas,omitempty"` + Host string `json:"host,omitempty"` + Affinity *core.Affinity `json:"affinity,omitempty"` + Tolerations []core.Toleration `json:"tolerations,omitempty"` // Metadata applied to the Deployment pod template. Metadata ApicurioRegistrySpecDeploymentMetadata `json:"metadata,omitempty"` // Image set in the Deployment pod template. Overrides the values in the REGISTRY_IMAGE_MEM, REGISTRY_IMAGE_KAFKASQL and REGISTRY_IMAGE_SQL operator environment variables. Image string `json:"image,omitempty"` // List of secrets in the same namespace to use for pulling the Deployment pod image. - ImagePullSecrets []corev1.LocalObjectReference `json:"imagePullSecrets,omitempty"` + ImagePullSecrets []core.LocalObjectReference `json:"imagePullSecrets,omitempty"` // Configure how the Operator manages Kubernetes resources ManagedResources ApicurioRegistrySpecDeploymentManagedResources `json:"managedResources,omitempty"` + + PodSpecPreview ApicurioRegistrySpecDeploymentPodSpecPreview `json:"podSpecPreview,omitempty"` +} + +// This is a slightly modified copy of k8s.io/api/core/v1.PodSpec: +// - Allow empty field "containers" +// - Comments removed to avoid an error "Too long: must have at most 262144 bytes" when executing "kubectl apply". +// By using kubectl apply to create/update resources, an annotation "kubectl.kubernetes.io/last-applied-configuration" +// is created by K8s API to store the latest version of the resource. +// However, it has a size limit and if the CRD has many long descriptions, it will result the error. +type ApicurioRegistrySpecDeploymentPodSpecPreview struct { + Volumes []core.Volume `json:"volumes,omitempty" patchStrategy:"merge,retainKeys" patchMergeKey:"name" protobuf:"bytes,1,rep,name=volumes"` + + InitContainers []core.Container `json:"initContainers,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,20,rep,name=initContainers"` + + Containers []ApicurioRegistrySpecDeploymentPodSpecPreviewContainer `json:"containers,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,2,rep,name=containers"` + + EphemeralContainers []core.EphemeralContainer `json:"ephemeralContainers,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,34,rep,name=ephemeralContainers"` + + RestartPolicy core.RestartPolicy `json:"restartPolicy,omitempty" protobuf:"bytes,3,opt,name=restartPolicy,casttype=RestartPolicy"` + + TerminationGracePeriodSeconds *int64 `json:"terminationGracePeriodSeconds,omitempty" protobuf:"varint,4,opt,name=terminationGracePeriodSeconds"` + + ActiveDeadlineSeconds *int64 `json:"activeDeadlineSeconds,omitempty" protobuf:"varint,5,opt,name=activeDeadlineSeconds"` + + DNSPolicy core.DNSPolicy `json:"dnsPolicy,omitempty" protobuf:"bytes,6,opt,name=dnsPolicy,casttype=DNSPolicy"` + + NodeSelector map[string]string `json:"nodeSelector,omitempty" protobuf:"bytes,7,rep,name=nodeSelector"` + + ServiceAccountName string `json:"serviceAccountName,omitempty" protobuf:"bytes,8,opt,name=serviceAccountName"` + + DeprecatedServiceAccount string `json:"serviceAccount,omitempty" protobuf:"bytes,9,opt,name=serviceAccount"` + + AutomountServiceAccountToken *bool `json:"automountServiceAccountToken,omitempty" protobuf:"varint,21,opt,name=automountServiceAccountToken"` + + NodeName string `json:"nodeName,omitempty" protobuf:"bytes,10,opt,name=nodeName"` + + HostNetwork bool `json:"hostNetwork,omitempty" protobuf:"varint,11,opt,name=hostNetwork"` + + HostPID bool `json:"hostPID,omitempty" protobuf:"varint,12,opt,name=hostPID"` + + HostIPC bool `json:"hostIPC,omitempty" protobuf:"varint,13,opt,name=hostIPC"` + + ShareProcessNamespace *bool `json:"shareProcessNamespace,omitempty" protobuf:"varint,27,opt,name=shareProcessNamespace"` + + SecurityContext *core.PodSecurityContext `json:"securityContext,omitempty" protobuf:"bytes,14,opt,name=securityContext"` + + ImagePullSecrets []core.LocalObjectReference `json:"imagePullSecrets,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,15,rep,name=imagePullSecrets"` + + Hostname string `json:"hostname,omitempty" protobuf:"bytes,16,opt,name=hostname"` + + Subdomain string `json:"subdomain,omitempty" protobuf:"bytes,17,opt,name=subdomain"` + + Affinity *core.Affinity `json:"affinity,omitempty" protobuf:"bytes,18,opt,name=affinity"` + + SchedulerName string `json:"schedulerName,omitempty" protobuf:"bytes,19,opt,name=schedulerName"` + + Tolerations []core.Toleration `json:"tolerations,omitempty" protobuf:"bytes,22,opt,name=tolerations"` + + HostAliases []core.HostAlias `json:"hostAliases,omitempty" patchStrategy:"merge" patchMergeKey:"ip" protobuf:"bytes,23,rep,name=hostAliases"` + + PriorityClassName string `json:"priorityClassName,omitempty" protobuf:"bytes,24,opt,name=priorityClassName"` + + Priority *int32 `json:"priority,omitempty" protobuf:"bytes,25,opt,name=priority"` + + DNSConfig *core.PodDNSConfig `json:"dnsConfig,omitempty" protobuf:"bytes,26,opt,name=dnsConfig"` + + ReadinessGates []core.PodReadinessGate `json:"readinessGates,omitempty" protobuf:"bytes,28,opt,name=readinessGates"` + + RuntimeClassName *string `json:"runtimeClassName,omitempty" protobuf:"bytes,29,opt,name=runtimeClassName"` + + EnableServiceLinks *bool `json:"enableServiceLinks,omitempty" protobuf:"varint,30,opt,name=enableServiceLinks"` + + PreemptionPolicy *core.PreemptionPolicy `json:"preemptionPolicy,omitempty" protobuf:"bytes,31,opt,name=preemptionPolicy"` + + Overhead core.ResourceList `json:"overhead,omitempty" protobuf:"bytes,32,opt,name=overhead"` + + TopologySpreadConstraints []core.TopologySpreadConstraint `json:"topologySpreadConstraints,omitempty" patchStrategy:"merge" patchMergeKey:"topologyKey" protobuf:"bytes,33,opt,name=topologySpreadConstraints"` + + SetHostnameAsFQDN *bool `json:"setHostnameAsFQDN,omitempty" protobuf:"varint,35,opt,name=setHostnameAsFQDN"` + + OS *core.PodOS `json:"os,omitempty" protobuf:"bytes,36,opt,name=os"` +} + +// This is a slightly modified copy of k8s.io/api/core/v1.Container: +// - Allow empty field "name" +// - Comments removed, see ApicurioRegistrySpecDeploymentPodSpecPreview for an explanation +type ApicurioRegistrySpecDeploymentPodSpecPreviewContainer struct { + Name string `json:"name,omitempty" protobuf:"bytes,1,opt,name=name"` + + Image string `json:"image,omitempty" protobuf:"bytes,2,opt,name=image"` + + Command []string `json:"command,omitempty" protobuf:"bytes,3,rep,name=command"` + + Args []string `json:"args,omitempty" protobuf:"bytes,4,rep,name=args"` + + WorkingDir string `json:"workingDir,omitempty" protobuf:"bytes,5,opt,name=workingDir"` + + Ports []core.ContainerPort `json:"ports,omitempty" patchStrategy:"merge" patchMergeKey:"containerPort" protobuf:"bytes,6,rep,name=ports"` + + EnvFrom []core.EnvFromSource `json:"envFrom,omitempty" protobuf:"bytes,19,rep,name=envFrom"` + + Env []core.EnvVar `json:"env,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,7,rep,name=env"` + + Resources core.ResourceRequirements `json:"resources,omitempty" protobuf:"bytes,8,opt,name=resources"` + + VolumeMounts []core.VolumeMount `json:"volumeMounts,omitempty" patchStrategy:"merge" patchMergeKey:"mountPath" protobuf:"bytes,9,rep,name=volumeMounts"` + + VolumeDevices []core.VolumeDevice `json:"volumeDevices,omitempty" patchStrategy:"merge" patchMergeKey:"devicePath" protobuf:"bytes,21,rep,name=volumeDevices"` + + LivenessProbe *core.Probe `json:"livenessProbe,omitempty" protobuf:"bytes,10,opt,name=livenessProbe"` + + ReadinessProbe *core.Probe `json:"readinessProbe,omitempty" protobuf:"bytes,11,opt,name=readinessProbe"` + + StartupProbe *core.Probe `json:"startupProbe,omitempty" protobuf:"bytes,22,opt,name=startupProbe"` + + Lifecycle *core.Lifecycle `json:"lifecycle,omitempty" protobuf:"bytes,12,opt,name=lifecycle"` + + TerminationMessagePath string `json:"terminationMessagePath,omitempty" protobuf:"bytes,13,opt,name=terminationMessagePath"` + + TerminationMessagePolicy core.TerminationMessagePolicy `json:"terminationMessagePolicy,omitempty" protobuf:"bytes,20,opt,name=terminationMessagePolicy,casttype=TerminationMessagePolicy"` + + ImagePullPolicy core.PullPolicy `json:"imagePullPolicy,omitempty" protobuf:"bytes,14,opt,name=imagePullPolicy,casttype=PullPolicy"` + + SecurityContext *core.SecurityContext `json:"securityContext,omitempty" protobuf:"bytes,15,opt,name=securityContext"` + + Stdin bool `json:"stdin,omitempty" protobuf:"varint,16,opt,name=stdin"` + + StdinOnce bool `json:"stdinOnce,omitempty" protobuf:"varint,17,opt,name=stdinOnce"` + + TTY bool `json:"tty,omitempty" protobuf:"varint,18,opt,name=tty"` } type ApicurioRegistrySpecDeploymentManagedResources struct { @@ -129,7 +260,7 @@ type ApicurioRegistryStatus struct { // Information about the deployed application. Info ApicurioRegistryStatusInfo `json:"info,omitempty"` // List of status conditions. - Conditions []metav1.Condition `json:"conditions,omitempty"` + Conditions []meta.Condition `json:"conditions,omitempty"` // List of resources managed by this operator. ManagedResources []ApicurioRegistryStatusManagedResource `json:"managedResources,omitempty"` } @@ -150,8 +281,8 @@ type ApicurioRegistryStatusManagedResource struct { // +kubebuilder:object:root=true // +kubebuilder:subresource:status type ApicurioRegistry struct { - metav1.TypeMeta `json:",inline"` - metav1.ObjectMeta `json:"metadata,omitempty"` + meta.TypeMeta `json:",inline"` + meta.ObjectMeta `json:"metadata,omitempty"` Spec ApicurioRegistrySpec `json:"spec,omitempty"` Status ApicurioRegistryStatus `json:"status,omitempty"` @@ -160,9 +291,9 @@ type ApicurioRegistry struct { // ApicurioRegistryList contains a list of ApicurioRegistry // +kubebuilder:object:root=true type ApicurioRegistryList struct { - metav1.TypeMeta `json:",inline"` - metav1.ListMeta `json:"metadata,omitempty"` - Items []ApicurioRegistry `json:"items"` + meta.TypeMeta `json:",inline"` + meta.ListMeta `json:"metadata,omitempty"` + Items []ApicurioRegistry `json:"items"` } func init() { diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go index fb945dce..3b39b613 100644 --- a/api/v1/zz_generated.deepcopy.go +++ b/api/v1/zz_generated.deepcopy.go @@ -307,6 +307,7 @@ func (in *ApicurioRegistrySpecDeployment) DeepCopyInto(out *ApicurioRegistrySpec copy(*out, *in) } out.ManagedResources = in.ManagedResources + in.PodSpecPreview.DeepCopyInto(&out.PodSpecPreview) } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ApicurioRegistrySpecDeployment. @@ -363,6 +364,241 @@ func (in *ApicurioRegistrySpecDeploymentMetadata) DeepCopy() *ApicurioRegistrySp return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ApicurioRegistrySpecDeploymentPodSpecPreview) DeepCopyInto(out *ApicurioRegistrySpecDeploymentPodSpecPreview) { + *out = *in + if in.Volumes != nil { + in, out := &in.Volumes, &out.Volumes + *out = make([]corev1.Volume, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.InitContainers != nil { + in, out := &in.InitContainers, &out.InitContainers + *out = make([]corev1.Container, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.Containers != nil { + in, out := &in.Containers, &out.Containers + *out = make([]ApicurioRegistrySpecDeploymentPodSpecPreviewContainer, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.EphemeralContainers != nil { + in, out := &in.EphemeralContainers, &out.EphemeralContainers + *out = make([]corev1.EphemeralContainer, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.TerminationGracePeriodSeconds != nil { + in, out := &in.TerminationGracePeriodSeconds, &out.TerminationGracePeriodSeconds + *out = new(int64) + **out = **in + } + if in.ActiveDeadlineSeconds != nil { + in, out := &in.ActiveDeadlineSeconds, &out.ActiveDeadlineSeconds + *out = new(int64) + **out = **in + } + if in.NodeSelector != nil { + in, out := &in.NodeSelector, &out.NodeSelector + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } + if in.AutomountServiceAccountToken != nil { + in, out := &in.AutomountServiceAccountToken, &out.AutomountServiceAccountToken + *out = new(bool) + **out = **in + } + if in.ShareProcessNamespace != nil { + in, out := &in.ShareProcessNamespace, &out.ShareProcessNamespace + *out = new(bool) + **out = **in + } + if in.SecurityContext != nil { + in, out := &in.SecurityContext, &out.SecurityContext + *out = new(corev1.PodSecurityContext) + (*in).DeepCopyInto(*out) + } + if in.ImagePullSecrets != nil { + in, out := &in.ImagePullSecrets, &out.ImagePullSecrets + *out = make([]corev1.LocalObjectReference, len(*in)) + copy(*out, *in) + } + if in.Affinity != nil { + in, out := &in.Affinity, &out.Affinity + *out = new(corev1.Affinity) + (*in).DeepCopyInto(*out) + } + if in.Tolerations != nil { + in, out := &in.Tolerations, &out.Tolerations + *out = make([]corev1.Toleration, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.HostAliases != nil { + in, out := &in.HostAliases, &out.HostAliases + *out = make([]corev1.HostAlias, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.Priority != nil { + in, out := &in.Priority, &out.Priority + *out = new(int32) + **out = **in + } + if in.DNSConfig != nil { + in, out := &in.DNSConfig, &out.DNSConfig + *out = new(corev1.PodDNSConfig) + (*in).DeepCopyInto(*out) + } + if in.ReadinessGates != nil { + in, out := &in.ReadinessGates, &out.ReadinessGates + *out = make([]corev1.PodReadinessGate, len(*in)) + copy(*out, *in) + } + if in.RuntimeClassName != nil { + in, out := &in.RuntimeClassName, &out.RuntimeClassName + *out = new(string) + **out = **in + } + if in.EnableServiceLinks != nil { + in, out := &in.EnableServiceLinks, &out.EnableServiceLinks + *out = new(bool) + **out = **in + } + if in.PreemptionPolicy != nil { + in, out := &in.PreemptionPolicy, &out.PreemptionPolicy + *out = new(corev1.PreemptionPolicy) + **out = **in + } + if in.Overhead != nil { + in, out := &in.Overhead, &out.Overhead + *out = make(corev1.ResourceList, len(*in)) + for key, val := range *in { + (*out)[key] = val.DeepCopy() + } + } + if in.TopologySpreadConstraints != nil { + in, out := &in.TopologySpreadConstraints, &out.TopologySpreadConstraints + *out = make([]corev1.TopologySpreadConstraint, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.SetHostnameAsFQDN != nil { + in, out := &in.SetHostnameAsFQDN, &out.SetHostnameAsFQDN + *out = new(bool) + **out = **in + } + if in.OS != nil { + in, out := &in.OS, &out.OS + *out = new(corev1.PodOS) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ApicurioRegistrySpecDeploymentPodSpecPreview. +func (in *ApicurioRegistrySpecDeploymentPodSpecPreview) DeepCopy() *ApicurioRegistrySpecDeploymentPodSpecPreview { + if in == nil { + return nil + } + out := new(ApicurioRegistrySpecDeploymentPodSpecPreview) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ApicurioRegistrySpecDeploymentPodSpecPreviewContainer) DeepCopyInto(out *ApicurioRegistrySpecDeploymentPodSpecPreviewContainer) { + *out = *in + if in.Command != nil { + in, out := &in.Command, &out.Command + *out = make([]string, len(*in)) + copy(*out, *in) + } + if in.Args != nil { + in, out := &in.Args, &out.Args + *out = make([]string, len(*in)) + copy(*out, *in) + } + if in.Ports != nil { + in, out := &in.Ports, &out.Ports + *out = make([]corev1.ContainerPort, len(*in)) + copy(*out, *in) + } + if in.EnvFrom != nil { + in, out := &in.EnvFrom, &out.EnvFrom + *out = make([]corev1.EnvFromSource, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.Env != nil { + in, out := &in.Env, &out.Env + *out = make([]corev1.EnvVar, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + in.Resources.DeepCopyInto(&out.Resources) + if in.VolumeMounts != nil { + in, out := &in.VolumeMounts, &out.VolumeMounts + *out = make([]corev1.VolumeMount, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.VolumeDevices != nil { + in, out := &in.VolumeDevices, &out.VolumeDevices + *out = make([]corev1.VolumeDevice, len(*in)) + copy(*out, *in) + } + if in.LivenessProbe != nil { + in, out := &in.LivenessProbe, &out.LivenessProbe + *out = new(corev1.Probe) + (*in).DeepCopyInto(*out) + } + if in.ReadinessProbe != nil { + in, out := &in.ReadinessProbe, &out.ReadinessProbe + *out = new(corev1.Probe) + (*in).DeepCopyInto(*out) + } + if in.StartupProbe != nil { + in, out := &in.StartupProbe, &out.StartupProbe + *out = new(corev1.Probe) + (*in).DeepCopyInto(*out) + } + if in.Lifecycle != nil { + in, out := &in.Lifecycle, &out.Lifecycle + *out = new(corev1.Lifecycle) + (*in).DeepCopyInto(*out) + } + if in.SecurityContext != nil { + in, out := &in.SecurityContext, &out.SecurityContext + *out = new(corev1.SecurityContext) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ApicurioRegistrySpecDeploymentPodSpecPreviewContainer. +func (in *ApicurioRegistrySpecDeploymentPodSpecPreviewContainer) DeepCopy() *ApicurioRegistrySpecDeploymentPodSpecPreviewContainer { + if in == nil { + return nil + } + out := new(ApicurioRegistrySpecDeploymentPodSpecPreviewContainer) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ApicurioRegistryStatus) DeepCopyInto(out *ApicurioRegistryStatus) { *out = *in diff --git a/controllers/apicurioregistry_controller.go b/controllers/apicurioregistry_controller.go index 0cd12d7f..7bd97a58 100644 --- a/controllers/apicurioregistry_controller.go +++ b/controllers/apicurioregistry_controller.go @@ -222,15 +222,8 @@ func (this *ApicurioRegistryReconciler) createNewLoop(appName c.Name, appNamespa //deployment result.AddControlFunction(cf.NewDeploymentCF(ctx, loopServices)) - //depends on deployment - if features.SupportsPDBv1beta1 { - result.AddControlFunction(cf.NewPodDisruptionBudgetV1beta1CF(ctx, loopServices)) - } - if features.SupportsPDBv1 { - result.AddControlFunction(cf.NewPodDisruptionBudgetV1CF(ctx, loopServices)) - } - //deployment modifiers + result.AddControlFunction(cf.NewPodSpecCF(ctx, loopServices)) result.AddControlFunction(cf.NewAffinityCF(ctx)) result.AddControlFunction(cf.NewTolerationCF(ctx)) result.AddControlFunction(cf.NewAnnotationsCF(ctx)) @@ -254,9 +247,18 @@ func (this *ApicurioRegistryReconciler) createNewLoop(appName c.Name, appNamespa //env vars applier result.AddControlFunction(cf.NewEnvApplyCF(ctx)) + //depends on deployment + if features.SupportsPDBv1beta1 { + result.AddControlFunction(cf.NewPodDisruptionBudgetV1beta1CF(ctx, loopServices)) + } + if features.SupportsPDBv1 { + result.AddControlFunction(cf.NewPodDisruptionBudgetV1CF(ctx, loopServices)) + } + //service result.AddControlFunction(cf.NewServiceCF(ctx, loopServices)) + // service modifiers result.AddControlFunction(cf.NewHttpsCF(ctx, loopServices)) // depends on service @@ -266,12 +268,12 @@ func (this *ApicurioRegistryReconciler) createNewLoop(appName c.Name, appNamespa //result.AddControlFunction(cf.NewServiceMonitorCF(ctx, loopServices)) } + // network policy + result.AddControlFunction(cf.NewNetworkPolicyCF(ctx, loopServices)) + // ingress result.AddControlFunction(cf.NewIngressCF(ctx, loopServices)) - //network policy - result.AddControlFunction(cf.NewNetworkPolicyCF(ctx, loopServices)) - //dependents of ingress if features.IsOCP { result.AddControlFunction(cf.NewHostInitRouteOcpCF(ctx)) diff --git a/controllers/cf/cf_https.go b/controllers/cf/cf_https.go index c7cb6b11..c6b04672 100644 --- a/controllers/cf/cf_https.go +++ b/controllers/cf/cf_https.go @@ -248,7 +248,7 @@ func (this *HttpsCF) Respond() { if this.previousSecretName != "" { common.RemoveVolumeFromDeployment(deployment, NewSecretVolume(this.previousSecretName)) } - common.SetVolumeInDeployment(deployment, NewSecretVolume(this.targetSecretName)) + common.SetVolumeInDeployment(this.log, deployment, NewSecretVolume(this.targetSecretName)) this.log.Debugw("added secret volume") } if !this.httpsEnabled && this.secretVolumeExists { diff --git a/controllers/cf/cf_image.go b/controllers/cf/cf_image.go index 47910187..ecd43348 100644 --- a/controllers/cf/cf_image.go +++ b/controllers/cf/cf_image.go @@ -110,6 +110,8 @@ func (this *ImageCF) Sense() { } func (this *ImageCF) Compare() bool { + this.log.Debugw("Observation", "this.existingImage", this.existingImage, + "this.targetImage", this.targetImage) // Condition #1 // Deployment exists // Condition #2 diff --git a/controllers/cf/cf_pod_spec.go b/controllers/cf/cf_pod_spec.go new file mode 100644 index 00000000..0de3952b --- /dev/null +++ b/controllers/cf/cf_pod_spec.go @@ -0,0 +1,318 @@ +package cf + +import ( + "encoding/json" + ar "github.com/Apicurio/apicurio-registry-operator/api/v1" + "github.com/Apicurio/apicurio-registry-operator/controllers/common" + "github.com/Apicurio/apicurio-registry-operator/controllers/loop" + "github.com/Apicurio/apicurio-registry-operator/controllers/loop/context" + "github.com/Apicurio/apicurio-registry-operator/controllers/loop/services" + "github.com/Apicurio/apicurio-registry-operator/controllers/svc/resources" + "go.uber.org/zap" + apps "k8s.io/api/apps/v1" + core "k8s.io/api/core/v1" + "reflect" +) + +var _ loop.ControlFunction = &PodSpecCF{} + +type PodSpecCF struct { + ctx context.LoopContext + log *zap.SugaredLogger + svcResourceCache resources.ResourceCache + services services.LoopServices + + previousBasePodSpec *ar.ApicurioRegistrySpecDeploymentPodSpecPreview + basePodSpec *ar.ApicurioRegistrySpecDeploymentPodSpecPreview + valid bool + targetPodSpec *core.PodSpec +} + +func NewPodSpecCF(ctx context.LoopContext, services services.LoopServices) loop.ControlFunction { + res := &PodSpecCF{ + ctx: ctx, + svcResourceCache: ctx.GetResourceCache(), + services: services, + } + res.log = ctx.GetLog().Sugar().With("cf", res.Describe()) + return res +} + +func (this *PodSpecCF) Describe() string { + return "PodSpecCF" +} + +func (this *PodSpecCF) Sense() { + this.valid = false + + if entry, exists := this.svcResourceCache.Get(resources.RC_KEY_SPEC); exists { + + this.basePodSpec = &entry.GetValue().(*ar.ApicurioRegistry).Spec.Deployment.PodSpecPreview + this.basePodSpec = this.basePodSpec.DeepCopy() // Defensive copy so we don't update the spec + + if deploymentEntry, deploymentExists := this.svcResourceCache.Get(resources.RC_KEY_DEPLOYMENT); deploymentExists { + currentPodSpec := &deploymentEntry.GetValue().(*apps.Deployment).Spec.Template.Spec + currentPodSpec = currentPodSpec.DeepCopy() // TODO? + factoryPodSpec := this.services.GetKubeFactory().CreateDeployment().Spec.Template.Spec // TODO Cache this? + targetPodSpec, err := SanitizeBasePodSpec(this.log, this.basePodSpec, currentPodSpec, &factoryPodSpec) + if err != nil { + if err.isConfigError { + this.log.Errorw("PodSpec field in spec.deployment.podSpecPreview is invalid", "error", err) + this.services.GetConditionManager().GetConfigurationErrorCondition(). + TransitionInvalid(err.Error(), "spec.deployment.podSpecPreview") + // No need to transition to not ready, since we can just with the previous config + this.ctx.SetRequeueDelaySec(10) + } else { + this.log.Errorw("Conversion error", "error", err.conversionError) + } + } else { + + this.targetPodSpec, err = ConvertToPodSpec(targetPodSpec) + if err != nil { + this.log.Errorw("Conversion error", "error", err.conversionError) + } else { + this.valid = true + } + } + } + } +} + +func (this *PodSpecCF) Compare() bool { + if this.previousBasePodSpec != nil { // common.LogNillable does not work TODO find out why + this.log.Debugw("Obsevation #1", "this.previousBasePodSpec", this.previousBasePodSpec) + } else { + this.log.Debugw("Obsevation #1", "this.previousBasePodSpec", "") + } + this.log.Debugw("Obsevation #2", "this.basePodSpec", this.basePodSpec) + this.log.Debugw("Obsevation #3", "this.targetPodSpec", this.targetPodSpec) + return this.valid && + // We're only comparing changes to the podSpecPreview, not the real pod spec, + // so we do not overwrite changes by the other CFs, which would cause a loop panic + (this.previousBasePodSpec == nil || !reflect.DeepEqual(this.basePodSpec, this.previousBasePodSpec)) +} + +func (this *PodSpecCF) Respond() { + + if entry, exists := this.svcResourceCache.Get(resources.RC_KEY_DEPLOYMENT); exists { + entry.ApplyPatch(func(value interface{}) interface{} { + deployment := value.(*apps.Deployment).DeepCopy() + + deployment.Spec.Template.Spec = *this.targetPodSpec + + this.previousBasePodSpec = this.basePodSpec + + return deployment + }) + } +} + +func (this *PodSpecCF) Cleanup() bool { + // No cleanup + return true +} + +/* +Not allowed: + +- affinity [alternative exists] +- containers[*]. [only a single container without a name] + - containers.args [reserved] + - containers.command [reserved] + - containers.env [alternative exists] + - containers.image [reserved] + - containers.imagePullPolicy [alternative exists] + - containers.name [reserved] + - containers.ports [reserved] + - containers.workingDir [reserved] +- ephemeralContainers [reserved] +- imagePullSecrets [alternative exists] +- initContainers [reserved] +- tolerations [alternative exists] + +Rules: + +If you are setting a value of a field in the podSpec or podSpec.containers[0], +this value must be valid as a whole. + +For example, when setting the initialDelaySeconds field of a readiness probe, you must also provide the probe handler. + +The operator may still modify the values you provided, +but it will not add required sub-fields or fix an invalid value. +*/ +func SanitizeBasePodSpec(log *zap.SugaredLogger, base *ar.ApicurioRegistrySpecDeploymentPodSpecPreview, current *core.PodSpec, factory *core.PodSpec) (*ar.ApicurioRegistrySpecDeploymentPodSpecPreview, *SanitizeError) { + // We are using values from *current* with fields (values): + // - that the user cannot change + // - or are empty by default + // - or there is a CF that handles them + // We are using values from *factory* with fields (values): + // - that are not empty by default, so empty fields in the base podSpec do not remove default values + // - and there is not and existing CF that handles them + // Otherwise we just pass the base values along, relying on CFs to override things + + base = base.DeepCopy() // Defensive copy TODO is this needed? + current = current.DeepCopy() // Defensive copy TODO is this needed? + // affinity + if base.Affinity != nil { + return nil, NewConfigError("affinity") + } + base.Affinity = current.Affinity + // containers[*] + if len(base.Containers) > 1 { + return nil, NewConfigError("field containers must contain at most one item") + + } else if len(base.Containers) == 1 { + baseContainer := &base.Containers[0] + // We have only a single container at the moment, + // TODO but in the future we should use the name field. + currentContainer := current.Containers[0] + factoryContainer := factory.Containers[0] + + // containers[*].args + if len(baseContainer.Args) > 0 { + return nil, NewConfigError("containers[0].args") + } + baseContainer.Args = currentContainer.Args + // containers[*].command + if len(baseContainer.Command) > 0 { + return nil, NewConfigError("containers[0].command") + } + baseContainer.Command = currentContainer.Command + // containers[*].env + if len(baseContainer.Env) > 0 { + return nil, NewConfigError("containers[0].env") + } + baseContainer.Env = currentContainer.Env + // containers[*].image + if baseContainer.Image != "" { + return nil, NewConfigError("containers[0].image") + } + baseContainer.Image = currentContainer.Image + // containers[*].imagePullPolicy + if baseContainer.ImagePullPolicy != "" { + return nil, NewConfigError("containers[0].imagePullPolicy") + } + baseContainer.ImagePullPolicy = currentContainer.ImagePullPolicy + // containers[*].livenessProbe + if baseContainer.LivenessProbe == nil { + baseContainer.LivenessProbe = factoryContainer.LivenessProbe + } + // containers[*].name + if baseContainer.Name != "" { + return nil, NewConfigError("containers[0].name") + } + baseContainer.Name = factoryContainer.Name + // containers[*].ports + if len(baseContainer.Ports) > 0 { + return nil, NewConfigError("containers[0].ports") + } + baseContainer.Ports = currentContainer.Ports + // containers[*].readinessProbe + if baseContainer.ReadinessProbe == nil { + baseContainer.ReadinessProbe = factoryContainer.ReadinessProbe + } + // containers[*].resources + if len(baseContainer.Resources.Limits) == 0 && + len(baseContainer.Resources.Requests) == 0 { + baseContainer.Resources = factoryContainer.Resources + } + // containers[*].workingDir + if baseContainer.WorkingDir != "" { + return nil, NewConfigError("containers[0].workingDir") + } + baseContainer.WorkingDir = currentContainer.WorkingDir + } else { + base.Containers = make([]ar.ApicurioRegistrySpecDeploymentPodSpecPreviewContainer, len(factory.Containers)) + for i, c := range factory.Containers { + cc, err := ConvertFromContainer(&c) + if err != nil { + return nil, err + } + base.Containers[i] = *cc + } + } + // ephemeralContainers + if len(base.EphemeralContainers) > 0 { + return nil, NewConfigError("ephemeralContainers") + } + base.EphemeralContainers = current.EphemeralContainers + // imagePullSecrets + if len(base.ImagePullSecrets) > 0 { + return nil, NewConfigError("imagePullSecrets") + } + base.ImagePullSecrets = current.ImagePullSecrets + // initContainers + if len(base.InitContainers) > 0 { + return nil, NewConfigError("initContainers") + } + base.InitContainers = current.InitContainers + // terminationGracePeriodSeconds + if base.TerminationGracePeriodSeconds == nil { + base.TerminationGracePeriodSeconds = factory.TerminationGracePeriodSeconds + } + // tolerations + if len(base.Tolerations) > 0 { + return nil, NewConfigError("tolerations") + } + base.Tolerations = current.Tolerations + // volumes + for _, v := range factory.Volumes { + common.SetVolume(log, &base.Volumes, &v) + } + return base, nil +} + +type SanitizeError struct { + isConfigError bool + conversionError error + message string +} + +func (this *SanitizeError) Error() string { + return this.message +} + +func NewConfigError(field string) *SanitizeError { + return &SanitizeError{ + isConfigError: true, + conversionError: nil, + message: "field " + field + " is reserved and must not be defined", + } +} + +func NewConvertError(err error) *SanitizeError { + return &SanitizeError{ + isConfigError: false, + conversionError: err, + message: err.Error(), + } +} + +// Do a magic using JSON to conver these values. +// They MUST have the equivalent field names. +// Remember, we are only doing this because we don't have ommitempty tags in PodSpec. +func ConvertToPodSpec(source *ar.ApicurioRegistrySpecDeploymentPodSpecPreview) (*core.PodSpec, *SanitizeError) { + data, err := json.Marshal(source) + if err != nil { + return nil, NewConvertError(err) + } + out := &core.PodSpec{} + err = json.Unmarshal(data, out) + if err != nil { + return nil, NewConvertError(err) + } + return out, nil +} + +func ConvertFromContainer(source *core.Container) (*ar.ApicurioRegistrySpecDeploymentPodSpecPreviewContainer, *SanitizeError) { + data, err := json.Marshal(source) + if err != nil { + return nil, NewConvertError(err) + } + out := &ar.ApicurioRegistrySpecDeploymentPodSpecPreviewContainer{} + err = json.Unmarshal(data, out) + if err != nil { + return nil, NewConvertError(err) + } + return out, nil +} diff --git a/controllers/common/util.go b/controllers/common/util.go index 8f7e4ba8..16fb9dc1 100644 --- a/controllers/common/util.go +++ b/controllers/common/util.go @@ -104,21 +104,24 @@ func AssertSliceContains(t *testing.T, haystack []interface{}, needle interface{ } } -func SetVolumeInDeployment(deployment *apps.Deployment, volume *core.Volume) { +// TODO Refactor +func SetVolumeInDeployment(log *zap.SugaredLogger, deployment *apps.Deployment, volume *core.Volume) { + SetVolume(log, &deployment.Spec.Template.Spec.Volumes, volume) +} - deploymentVolumes := &deployment.Spec.Template.Spec.Volumes +func SetVolume(log *zap.SugaredLogger, volumes *[]core.Volume, volume *core.Volume) { volumeAlreadyExists := false // Modify volume if it exists, otherwise append as a new volume - for i, vol := range *deploymentVolumes { + for i, vol := range *volumes { if vol.Name == volume.Name { volumeAlreadyExists = true - (*deploymentVolumes)[i] = *volume + (*volumes)[i] = *volume break } } if !volumeAlreadyExists { - *deploymentVolumes = append(*deploymentVolumes, *volume) + *volumes = append(*volumes, *volume) } } diff --git a/controllers/svc/factory/factory_kube.go b/controllers/svc/factory/factory_kube.go index a7a10870..a0eb12ff 100644 --- a/controllers/svc/factory/factory_kube.go +++ b/controllers/svc/factory/factory_kube.go @@ -86,52 +86,53 @@ func (this *KubeFactory) CreateDeployment() *apps.Deployment { Labels: this.GetLabels(), }, Spec: core.PodSpec{ - Containers: []core.Container{{ - Name: this.ctx.GetAppName().Str(), - Resources: core.ResourceRequirements{ - Limits: core.ResourceList{ - core.ResourceCPU: resource.MustParse("1"), - core.ResourceMemory: resource.MustParse("1300Mi"), - }, - Requests: core.ResourceList{ - core.ResourceCPU: resource.MustParse("500m"), - core.ResourceMemory: resource.MustParse("512Mi"), + Containers: []core.Container{ + { + Name: this.ctx.GetAppName().Str(), + Resources: core.ResourceRequirements{ + Limits: core.ResourceList{ + core.ResourceCPU: resource.MustParse("1"), + core.ResourceMemory: resource.MustParse("1300Mi"), + }, + Requests: core.ResourceList{ + core.ResourceCPU: resource.MustParse("500m"), + core.ResourceMemory: resource.MustParse("512Mi"), + }, }, - }, - LivenessProbe: &core.Probe{ - ProbeHandler: core.ProbeHandler{ - HTTPGet: &core.HTTPGetAction{ - Path: "/health/live", - Port: intstr.FromInt(8080), + LivenessProbe: &core.Probe{ + ProbeHandler: core.ProbeHandler{ + HTTPGet: &core.HTTPGetAction{ + Path: "/health/live", + Port: intstr.FromInt(8080), + }, }, + InitialDelaySeconds: 15, + TimeoutSeconds: 5, + PeriodSeconds: 10, + SuccessThreshold: 1, + FailureThreshold: 3, }, - InitialDelaySeconds: 15, - TimeoutSeconds: 5, - PeriodSeconds: 10, - SuccessThreshold: 1, - FailureThreshold: 3, - }, - ReadinessProbe: &core.Probe{ - ProbeHandler: core.ProbeHandler{ - HTTPGet: &core.HTTPGetAction{ - Path: "/health/ready", - Port: intstr.FromInt(8080), + ReadinessProbe: &core.Probe{ + ProbeHandler: core.ProbeHandler{ + HTTPGet: &core.HTTPGetAction{ + Path: "/health/ready", + Port: intstr.FromInt(8080), + }, }, + InitialDelaySeconds: 15, + TimeoutSeconds: 5, + PeriodSeconds: 10, + SuccessThreshold: 1, + FailureThreshold: 3, }, - InitialDelaySeconds: 15, - TimeoutSeconds: 5, - PeriodSeconds: 10, - SuccessThreshold: 1, - FailureThreshold: 3, - }, - TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []core.VolumeMount{ - { - Name: "tmp", - MountPath: "/tmp", + VolumeMounts: []core.VolumeMount{ + { + Name: "tmp", + MountPath: "/tmp", + }, }, }, - }}, + }, TerminationGracePeriodSeconds: &terminationGracePeriodSeconds, Volumes: []core.Volume{ { diff --git a/controllers/svc/status/conditions/condition_configuration_error.go b/controllers/svc/status/conditions/condition_configuration_error.go index 85f9588c..c8a4dbbb 100644 --- a/controllers/svc/status/conditions/condition_configuration_error.go +++ b/controllers/svc/status/conditions/condition_configuration_error.go @@ -38,12 +38,12 @@ func (this *ConfigurationErrorCondition) TransitionRequired(optionPath string) { } } -func (this *ConfigurationErrorCondition) TransitionInvalid(currentValue string, optionPath string) { +func (this *ConfigurationErrorCondition) TransitionInvalid(details string, optionPath string) { if this.data.Reason != string(CONFIGURATION_ERROR_CONDITION_REASON_INVALID_PERSISTENCE) && this.data.Reason != string(CONFIGURATION_ERROR_CONDITION_REASON_REQUIRED) { this.data.Status = metav1.ConditionTrue this.data.Reason = string(CONFIGURATION_ERROR_CONDITION_REASON_INVALID) - this.data.Message = "Invalid value for configuration option " + optionPath + ": " + currentValue + this.data.Message = "Invalid value for configuration option " + optionPath + ": " + details } }