Allow drop slots when it gets deleted from the manifest

Author: idanovinda

pkg/cluster/sync.go

@@ -541,6 +541,45 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectiv
 	configPatched := false
 	requiresMasterRestart := false
 
+	for slotName, _ := range effectivePatroniConfig.Slots {
+		if _, exists := desiredPatroniConfig.Slots[slotName]; exists {
+			continue
+		}
+
+		configToRewrite := make(map[string]interface{})
+		configToRewrite["loop_wait"] = effectivePatroniConfig.LoopWait
+		configToRewrite["maximum_lag_on_failover"] = effectivePatroniConfig.MaximumLagOnFailover
+		configToRewrite["pg_hba"] = effectivePatroniConfig.PgHba
+		configToRewrite["retry_timeout"] = effectivePatroniConfig.RetryTimeout
+		configToRewrite["synchronous_mode"] = effectivePatroniConfig.SynchronousMode
+		configToRewrite["synchronous_mode_strict"] = effectivePatroniConfig.SynchronousModeStrict
+		configToRewrite["ttl"] = effectivePatroniConfig.TTL
+		configToRewrite["postgresql"] = map[string]interface{}{constants.PatroniPGParametersParameterName: effectivePgParameters}
+
+		slotsToRewrite := make(map[string]map[string]string)
+		for slotName, desiredSlot := range desiredPatroniConfig.Slots {
+			slotsToRewrite[slotName] = desiredSlot
+		}
+
+		if len(slotsToRewrite) > 0 {
+			configToRewrite["slots"] = slotsToRewrite
+		}
+
+		configToRewriteJson, err := json.Marshal(configToRewrite)
+		if err != nil {
+			c.logger.Debugf("could not convert config rewrite to JSON: %v", err)
+		}
+
+		podName := util.NameFromMeta(pod.ObjectMeta)
+		c.logger.Debugf("rewrite Postgres config via Patroni API on pod %s with following options: %s",
+			podName, configToRewriteJson)
+		if err = c.patroni.RewriteConfig(pod, configToRewrite); err != nil {
+			return configPatched, requiresMasterRestart, fmt.Errorf("could not rewrite postgres parameters within pod %s: %v", podName, err)
+		}
+
+		break
+	}
+
 	// compare effective and desired Patroni config options
 	if desiredPatroniConfig.LoopWait > 0 && desiredPatroniConfig.LoopWait != effectivePatroniConfig.LoopWait {
 		configToSet["loop_wait"] = desiredPatroniConfig.LoopWait

pkg/util/patroni/patroni.go

@@ -38,6 +38,7 @@ type Interface interface {
 	Restart(server *v1.Pod) error
 	GetConfig(server *v1.Pod) (acidv1.Patroni, map[string]string, error)
 	SetConfig(server *v1.Pod, config map[string]interface{}) error
+	RewriteConfig(server *v1.Pod, config map[string]interface{}) error
 }
 
 // Patroni API client
@@ -113,6 +114,42 @@ func (p *Patroni) httpPostOrPatch(method string, url string, body *bytes.Buffer)
 	return nil
 }
 
+func (p *Patroni) httpPut(method string, url string, body *bytes.Buffer) (err error) {
+	request, err := http.NewRequest(method, url, body)
+	if err != nil {
+		return fmt.Errorf("could not create request: %v", err)
+	}
+
+	if p.logger != nil {
+		p.logger.Debugf("making %s http request: %s", method, request.URL.String())
+	}
+
+	resp, err := p.httpClient.Do(request)
+	if err != nil {
+		return fmt.Errorf("could not make request: %v", err)
+	}
+	defer func() {
+		if err2 := resp.Body.Close(); err2 != nil {
+			if err != nil {
+				err = fmt.Errorf("could not close request: %v, prior error: %v", err2, err)
+			} else {
+				err = fmt.Errorf("could not close request: %v", err2)
+			}
+			return
+		}
+	}()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return fmt.Errorf("could not read response: %v", err)
+		}
+
+		return fmt.Errorf("patroni returned '%s'", string(bodyBytes))
+	}
+	return nil
+}
+
 func (p *Patroni) httpGet(url string) (string, error) {
 	p.logger.Debugf("making GET http request: %s", url)
 
@@ -178,6 +215,19 @@ func (p *Patroni) SetConfig(server *v1.Pod, config map[string]interface{}) error
 	return p.httpPostOrPatch(http.MethodPatch, apiURLString+configPath, buf)
 }
 
+func (p *Patroni) RewriteConfig(server *v1.Pod, config map[string]interface{}) error {
+	buf := &bytes.Buffer{}
+	err := json.NewEncoder(buf).Encode(config)
+	if err != nil {
+		return fmt.Errorf("could not encode json: %v", err)
+	}
+	apiURLString, err := apiURL(server)
+	if err != nil {
+		return err
+	}
+	return p.httpPut(http.MethodPut, apiURLString+configPath, buf)
+}
+
 // ClusterMembers array of cluster members from Patroni API
 type ClusterMembers struct {
 	Members []ClusterMember `json:"members"`