public_git
/
postgres-operator
mirror of https://github.com/zalando/postgres-operator.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

changes to postgresql CR status 

Signed-off-by: RavinaChidambaram  <cravina.19@gmail.com>
This commit is contained in:
RavinaChidambaram 2024-08-02 13:32:46 +05:30
parent 31f474a95c
commit d89e726b56
9 changed files with 243 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
manifests/minimal-postgres-manifest.yaml
View File

@ -2,6 +2,8 @@ apiVersion: "acid.zalan.do/v1"
kind: postgresql
metadata:
name: acid-minimal-cluster
labels:
cluster-name: acid-minimal-cluster
spec:
teamId: "acid"
volume:

35
manifests/postgresql.crd.yaml
View File

@ -20,6 +20,10 @@ spec:
storage: true
subresources:
status: {}
scale:
specReplicasPath: .spec.numberOfInstances
statusReplicasPath: .status.numberOfInstances
labelSelectorPath: .status.labelSelector
additionalPrinterColumns:
- name: Team
type: string
@ -51,7 +55,7 @@ spec:
- name: Status
type: string
description: Current sync status of postgresql resource
jsonPath: .status.PostgresClusterStatus
jsonPath: .status.postgresClusterStatus
schema:
openAPIV3Schema:
type: object
@ -677,5 +681,30 @@ spec:
type: integer
status:
type: object
additionalProperties:
type: string
properties:
postgresClusterStatus:
type: string
numberOfInstances:
format: int32
type: integer
labelSelector:
type: string
observedGeneration:
format: int64
type: integer
conditions:
type: array
items:
type: object
properties:
type:
type: string
status:
type: string
lastTransitionTime:
type: string
format: date-time
reason:
type: string
message:
type: string

66
pkg/apis/acid.zalan.do/v1/crds.go
View File

@ -25,6 +25,12 @@ const (
OperatorConfigCRDResourceShort = "opconfig"
)
var (
specReplicasPath = ".spec.numberOfInstances"
statusReplicasPath = ".status.numberOfInstances"
labelSelectorPath = ".status.labelSelector"
)
// PostgresCRDResourceColumns definition of AdditionalPrinterColumns for postgresql CRD
var PostgresCRDResourceColumns = []apiextv1.CustomResourceColumnDefinition{
{
@ -72,7 +78,7 @@ var PostgresCRDResourceColumns = []apiextv1.CustomResourceColumnDefinition{
Name: "Status",
Type: "string",
Description: "Current sync status of postgresql resource",
JSONPath: ".status.PostgresClusterStatus",
JSONPath: ".status.postgresClusterStatus",
},
}
@ -1106,10 +1112,47 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
},
"status": {
Type: "object",
AdditionalProperties: &apiextv1.JSONSchemaPropsOrBool{
Schema: &apiextv1.JSONSchemaProps{
Properties: map[string]apiextv1.JSONSchemaProps{
"postgresClusterStatus": {
Type: "string",
},
"numberOfInstances": {
Type: "integer",
Format: "int32",
},
"labelSelector": {
Type: "string",
},
"observedGeneration": {
Type: "integer",
Format: "int64",
},
"conditions": {
Type: "array",
Items: &apiextv1.JSONSchemaPropsOrArray{
Schema: &apiextv1.JSONSchemaProps{
Type: "object",
Properties: map[string]apiextv1.JSONSchemaProps{
"type": {
Type: "string",
},
"status": {
Type: "string",
},
"lastTransitionTime": {
Type: "string",
Format: "date-time",
},
"reason": {
Type: "string",
},
"message": {
Type: "string",
},
},
},
},
},
},
},
},
@ -1983,7 +2026,7 @@ var OperatorConfigCRDResourceValidation = apiextv1.CustomResourceValidation{
func buildCRD(name, kind, plural, list, short string,
categories []string,
columns []apiextv1.CustomResourceColumnDefinition,
validation apiextv1.CustomResourceValidation) *apiextv1.CustomResourceDefinition {
validation apiextv1.CustomResourceValidation, specReplicasPath string, statusReplicasPath string, labelSelectorPath string) *apiextv1.CustomResourceDefinition {
return &apiextv1.CustomResourceDefinition{
TypeMeta: metav1.TypeMeta{
APIVersion: fmt.Sprintf("%s/%s", apiextv1.GroupName, apiextv1.SchemeGroupVersion.Version),
@ -2010,6 +2053,11 @@ func buildCRD(name, kind, plural, list, short string,
Storage: true,
Subresources: &apiextv1.CustomResourceSubresources{
Status: &apiextv1.CustomResourceSubresourceStatus{},
Scale: &apiextv1.CustomResourceSubresourceScale{
SpecReplicasPath: specReplicasPath,
StatusReplicasPath: statusReplicasPath,
LabelSelectorPath: &labelSelectorPath,
},
},
AdditionalPrinterColumns: columns,
Schema: &validation,
@ -2028,7 +2076,10 @@ func PostgresCRD(crdCategories []string) *apiextv1.CustomResourceDefinition {
PostgresCRDResourceShort,
crdCategories,
PostgresCRDResourceColumns,
PostgresCRDResourceValidation)
PostgresCRDResourceValidation,
specReplicasPath,
statusReplicasPath,
labelSelectorPath)
}
// ConfigurationCRD returns CustomResourceDefinition built from OperatorConfigCRDResource
@ -2040,5 +2091,8 @@ func ConfigurationCRD(crdCategories []string) *apiextv1.CustomResourceDefinition
OperatorConfigCRDResourceShort,
crdCategories,
OperatorConfigCRDResourceColumns,
OperatorConfigCRDResourceValidation)
OperatorConfigCRDResourceValidation,
specReplicasPath,
statusReplicasPath,
labelSelectorPath)
}

42
pkg/apis/acid.zalan.do/v1/postgresql_type.go
View File

@ -4,6 +4,7 @@ package v1
import (
"time"
"k8s.io/apimachinery/pkg/api/equality"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@ -225,9 +226,48 @@ type Sidecar struct {
// UserFlags defines flags (such as superuser, nologin) that could be assigned to individual users
type UserFlags []string
type Conditions []Condition
type ConditionType string
type VolatileTime struct {
Inner metav1.Time `json:",inline"`
}
// MarshalJSON implements the json.Marshaler interface.
func (t VolatileTime) MarshalJSON() ([]byte, error) {
return t.Inner.MarshalJSON()
}
// UnmarshalJSON implements the json.Unmarshaller interface.
func (t *VolatileTime) UnmarshalJSON(b []byte) error {
return t.Inner.UnmarshalJSON(b)
}
func init() {
equality.Semantic.AddFunc(
// Always treat VolatileTime fields as equivalent.
func(VolatileTime, VolatileTime) bool {
return true
},
)
}
// Condition contains the conditions of the PostgreSQL cluster
type Condition struct {
Type ConditionType `json:"type" description:"type of status condition"`
Status v1.ConditionStatus `json:"status" description:"status of the condition, one of True, False, Unknown"`
LastTransitionTime VolatileTime `json:"lastTransitionTime,omitempty" description:"last time the condition transit from one status to another"`
Reason string `json:"reason,omitempty" description:"one-word CamelCase reason for the condition's last transition"`
Message string `json:"message,omitempty" description:"human-readable message indicating details about last transition"`
}
// PostgresStatus contains status of the PostgreSQL cluster (running, creation failed etc.)
type PostgresStatus struct {
PostgresClusterStatus string `json:"PostgresClusterStatus"`
PostgresClusterStatus string `json:"postgresClusterStatus"`
NumberOfInstances int32 `json:"numberOfInstances"`
LabelSelector string `json:"labelSelector"`
ObservedGeneration int64 `json:"observedGeneration,omitempty"`
Conditions Conditions `json:"conditions,omitempty"`
}
// ConnectionPooler Options for connection pooler

6
pkg/apis/acid.zalan.do/v1/util.go
View File

@ -101,6 +101,6 @@ func (postgresStatus PostgresStatus) Creating() bool {
return postgresStatus.PostgresClusterStatus == ClusterStatusCreating
}
func (postgresStatus PostgresStatus) String() string {
return postgresStatus.PostgresClusterStatus
}
//func (postgresStatus PostgresStatus) String() string {
// return postgresStatus.PostgresClusterStatus
//}

18
pkg/cluster/cluster.go
View File

@ -260,11 +260,13 @@ func (c *Cluster) Create() (err error) {
pgUpdatedStatus *acidv1.Postgresql
errStatus error
)
labelstring := fmt.Sprintf("%s=%s", "cluster-name", c.Postgresql.ObjectMeta.Labels["cluster-name"])
existingCondition := c.Postgresql.Status.Conditions
if err == nil {
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusRunning) //TODO: are you sure it's running?
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusRunning, c.Postgresql.Spec.NumberOfInstances, labelstring, c.Postgresql.Generation, existingCondition, "") //TODO: are you sure it's running?
} else {
c.logger.Warningf("cluster created failed: %v", err)
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusAddFailed)
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusAddFailed, 0, labelstring, 0, existingCondition, err.Error())
}
if errStatus != nil {
c.logger.Warningf("could not set cluster status: %v", errStatus)
@ -274,7 +276,9 @@ func (c *Cluster) Create() (err error) {
}
}()
pgCreateStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusCreating)
labelstring := fmt.Sprintf("%s=%s", "cluster-name", c.Postgresql.ObjectMeta.Labels["cluster-name"])
existingCondition := c.Postgresql.Status.Conditions
pgCreateStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusCreating, 0, labelstring, 0, existingCondition, "")
if err != nil {
return fmt.Errorf("could not set cluster status: %v", err)
}
@ -927,7 +931,9 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
c.mu.Lock()
defer c.mu.Unlock()
c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusUpdating)
labelstring := fmt.Sprintf("%s=%s", "cluster-name", c.Postgresql.ObjectMeta.Labels["cluster-name"])
existingCondition := c.Postgresql.Status.Conditions
c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusUpdating, c.Postgresql.Status.NumberOfInstances, labelstring, c.Postgresql.Status.ObservedGeneration, existingCondition, "")
c.setSpec(newSpec)
defer func() {
@ -936,9 +942,9 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
err error
)
if updateFailed {
pgUpdatedStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusUpdateFailed)
pgUpdatedStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusUpdateFailed, c.Postgresql.Status.NumberOfInstances, labelstring, c.Postgresql.Status.ObservedGeneration, existingCondition, err.Error())
} else {
pgUpdatedStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusRunning)
pgUpdatedStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusRunning, newSpec.Spec.NumberOfInstances, labelstring, c.Postgresql.Generation, existingCondition, "")
}
if err != nil {
c.logger.Warningf("could not set cluster status: %v", err)

6
pkg/cluster/sync.go
View File

@ -46,11 +46,13 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
pgUpdatedStatus *acidv1.Postgresql
errStatus error
)
labelstring := fmt.Sprintf("%s=%s", "cluster-name", c.Postgresql.ObjectMeta.Labels["cluster-name"])
existingCondition := c.Postgresql.Status.Conditions
if err != nil {
c.logger.Warningf("error while syncing cluster state: %v", err)
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusSyncFailed)
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusSyncFailed, newSpec.Status.NumberOfInstances, labelstring, c.Postgresql.Status.ObservedGeneration, existingCondition, errStatus.Error())
} else if !c.Status.Running() {
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusRunning)
pgUpdatedStatus, errStatus = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusRunning, newSpec.Spec.NumberOfInstances, labelstring, c.Postgresql.Generation, existingCondition, "")
}
if errStatus != nil {
c.logger.Warningf("could not set cluster status: %v", errStatus)

12
pkg/controller/postgresql.go
View File

@ -161,7 +161,9 @@ func (c *Controller) acquireInitialListOfClusters() error {
func (c *Controller) addCluster(lg *logrus.Entry, clusterName spec.NamespacedName, pgSpec *acidv1.Postgresql) (*cluster.Cluster, error) {
if c.opConfig.EnableTeamIdClusternamePrefix {
if _, err := acidv1.ExtractClusterName(clusterName.Name, pgSpec.Spec.TeamID); err != nil {
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusInvalid)
labelstring := fmt.Sprintf("%s=%s", "cluster-name", pgSpec.ObjectMeta.Labels["cluster-name"])
existingCondition := pgSpec.Status.Conditions
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusInvalid, pgSpec.Status.NumberOfInstances, labelstring, pgSpec.Status.ObservedGeneration, existingCondition, err.Error())
return nil, err
}
}
@ -472,16 +474,18 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *acidv1.
if clusterError != "" && eventType != EventDelete {
c.logger.WithField("cluster-name", clusterName).Debugf("skipping %q event for the invalid cluster: %s", eventType, clusterError)
labelstring := fmt.Sprintf("%s=%s", "cluster-name", informerNewSpec.ObjectMeta.Labels["cluster-name"])
existingCondition := informerNewSpec.Status.Conditions
switch eventType {
case EventAdd:
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusAddFailed)
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusAddFailed, informerNewSpec.Status.NumberOfInstances, labelstring, informerNewSpec.Status.ObservedGeneration, existingCondition, clusterError)
c.eventRecorder.Eventf(c.GetReference(informerNewSpec), v1.EventTypeWarning, "Create", "%v", clusterError)
case EventUpdate:
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusUpdateFailed)
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusUpdateFailed, informerNewSpec.Status.NumberOfInstances, labelstring, informerNewSpec.Status.ObservedGeneration, existingCondition, clusterError)
c.eventRecorder.Eventf(c.GetReference(informerNewSpec), v1.EventTypeWarning, "Update", "%v", clusterError)
default:
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusSyncFailed)
c.KubeClient.SetPostgresCRDStatus(clusterName, acidv1.ClusterStatusSyncFailed, informerNewSpec.Status.NumberOfInstances, labelstring, informerNewSpec.Status.ObservedGeneration, existingCondition, clusterError)
c.eventRecorder.Eventf(c.GetReference(informerNewSpec), v1.EventTypeWarning, "Sync", "%v", clusterError)
}

83
pkg/util/k8sutil/k8sutil.go
View File

@ -3,6 +3,7 @@ package k8sutil
import (
"context"
"fmt"
"time"
b64 "encoding/base64"
"encoding/json"
@ -192,10 +193,17 @@ func NewFromConfig(cfg *rest.Config) (KubernetesClient, error) {
}
// SetPostgresCRDStatus of Postgres cluster
func (client *KubernetesClient) SetPostgresCRDStatus(clusterName spec.NamespacedName, status string) (*apiacidv1.Postgresql, error) {
func (client *KubernetesClient) SetPostgresCRDStatus(clusterName spec.NamespacedName, status string, numberOfInstances int32, labelSelector string, observedGeneration int64, existingCondition apiacidv1.Conditions, message string) (*apiacidv1.Postgresql, error) {
var pg *apiacidv1.Postgresql
var pgStatus apiacidv1.PostgresStatus
pgStatus := apiacidv1.PostgresStatus{}
pgStatus.PostgresClusterStatus = status
pgStatus.NumberOfInstances = numberOfInstances
pgStatus.LabelSelector = labelSelector
pgStatus.ObservedGeneration = observedGeneration
newConditions := updateConditions(existingCondition, status, message)
pgStatus.Conditions = newConditions
patch, err := json.Marshal(struct {
PgStatus interface{} `json:"status"`
@ -217,6 +225,77 @@ func (client *KubernetesClient) SetPostgresCRDStatus(clusterName spec.Namespaced
return pg, nil
}
func updateConditions(existingConditions apiacidv1.Conditions, currentStatus string, message string) apiacidv1.Conditions {
now := apiacidv1.VolatileTime{Inner: metav1.NewTime(time.Now())}
var readyCondition, reconciliationCondition *apiacidv1.Condition
// Find existing conditions
for i := range existingConditions {
if existingConditions[i].Type == "Ready" {
readyCondition = &existingConditions[i]
} else if existingConditions[i].Type == "ReconciliationSuccessful" {
reconciliationCondition = &existingConditions[i]
}
}
// Initialize conditions if not present
switch currentStatus {
case "Creating":
if reconciliationCondition == nil {
existingConditions = append(existingConditions, apiacidv1.Condition{Type: "ReconciliationSuccessful"})
reconciliationCondition = &existingConditions[len(existingConditions)-1]
}
default:
if readyCondition == nil {
existingConditions = append(existingConditions, apiacidv1.Condition{Type: "Ready"})
readyCondition = &existingConditions[len(existingConditions)-1]
}
}
// Update Ready condition
switch currentStatus {
case "Running":
readyCondition.Status = v1.ConditionTrue
readyCondition.LastTransitionTime = now
case "CreateFailed":
readyCondition.Status = v1.ConditionFalse
readyCondition.LastTransitionTime = now
case "UpdateFailed", "SyncFailed", "Invalid":
if readyCondition.Status == v1.ConditionFalse {
readyCondition.LastTransitionTime = now
}
case "Updating":
// not updatinf time, just setting the status
if readyCondition.Status == v1.ConditionFalse {
readyCondition.Status = v1.ConditionFalse
} else {
readyCondition.Status = v1.ConditionTrue
}
}
// Update ReconciliationSuccessful condition
reconciliationCondition.LastTransitionTime = now
reconciliationCondition.Message = message
if currentStatus == "Running" {
reconciliationCondition.Status = v1.ConditionTrue
reconciliationCondition.Reason = ""
} else {
reconciliationCondition.Status = v1.ConditionFalse
reconciliationCondition.Reason = currentStatus
}
// Directly modify elements in the existingConditions slice
for i := range existingConditions {
if existingConditions[i].Type == "Ready" && readyCondition != nil {
existingConditions[i] = *readyCondition
} else if existingConditions[i].Type == "ReconciliationSuccessful" && reconciliationCondition != nil {
existingConditions[i] = *reconciliationCondition
}
}
return existingConditions
}
// SetFinalizer of Postgres cluster
func (client *KubernetesClient) SetFinalizer(clusterName spec.NamespacedName, pg *apiacidv1.Postgresql, finalizers []string) (*apiacidv1.Postgresql, error) {
var (