Skip to content

Commit d224c90

Browse files
author
jmccormick2001
committed
part 2 of manual failover support, not finished
1 parent 436f02b commit d224c90

File tree

10 files changed

+516
-35
lines changed

10 files changed

+516
-35
lines changed

‎controller/taskcontroller.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ import (
3232

3333
// PgtaskController holds connections for the controller
3434
type PgtaskController struct {
35+
PgtaskConfig *rest.Config
3536
PgtaskClient *rest.RESTClient
3637
PgtaskScheme *runtime.Scheme
3738
PgtaskClientset *kubernetes.Clientset
@@ -102,7 +103,7 @@ func (c *PgtaskController) onAdd(obj interface{}) {
102103
return
103104
}
104105

105-
//update the status of the task as completed
106+
//update the status of the task as processed to prevent reprocessing
106107
taskCopy := copyObj.(*crv1.Pgtask)
107108
taskCopy.Status = crv1.PgtaskStatus{
108109
State: crv1.PgtaskStateProcessed,
@@ -129,7 +130,7 @@ func (c *PgtaskController) onAdd(obj interface{}) {
129130
log.Info("failover task added")
130131
log.Info("cluster name is " + task.Spec.Parameters)
131132
log.Info("dbname is " + task.Spec.Name)
132-
clusteroperator.FailoverBase(task.ObjectMeta.Namespace, c.PgtaskClientset, c.PgtaskClient, task)
133+
clusteroperator.FailoverBase(task.ObjectMeta.Namespace, c.PgtaskClientset, c.PgtaskClient, task, c.PgtaskConfig)
133134

134135
case crv1.PgtaskDeleteData:
135136
log.Info("delete data task added")
@@ -140,7 +141,7 @@ func (c *PgtaskController) onAdd(obj interface{}) {
140141
log.Info("unknown task type on pgtask added")
141142
}
142143

143-
//for now, remove the pgtask in all cases
144+
/**
144145
err = c.PgtaskClient.Delete().
145146
Name(task.ObjectMeta.Name).
146147
Namespace(task.ObjectMeta.Namespace).
@@ -154,6 +155,7 @@ func (c *PgtaskController) onAdd(obj interface{}) {
154155
} else {
155156
log.Errorf("deleted pgtask %s\n", task.ObjectMeta.Name)
156157
}
158+
*/
157159

158160
}
159161

‎golang-examples/addlabel.go‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -4,20 +4,21 @@ import (
44
"flag"
55
"fmt"
66
jsonpatch "github.com/evanphx/json-patch"
7+
meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
78

89
"k8s.io/client-go/kubernetes"
9-
"k8s.io/client-go/pkg/api"
10-
"k8s.io/client-go/pkg/util/json"
10+
//"k8s.io/client-go/pkg/api"
11+
"k8s.io/apimachinery/pkg/types"
1112
//"k8s.io/client-go/pkg/api/errors"
12-
"k8s.io/client-go/pkg/apis/extensions/v1beta1"
13+
"encoding/json"
14+
"k8s.io/api/extensions/v1beta1"
1315
//"k8s.io/client-go/pkg/runtime"
1416
//"k8s.io/client-go/pkg/runtime/serializer"
1517

1618
//"k8s.io/client-go/pkg/api/unversioned"
1719
//"k8s.io/client-go/pkg/api/v1"
1820
//"k8s.io/client-go/rest"
19-
//"k8s.io/apimachinery/pkg/api/meta"
20-
"k8s.io/client-go/pkg/api/meta"
21+
"k8s.io/apimachinery/pkg/api/meta"
2122

2223
"k8s.io/client-go/tools/clientcmd"
2324
)
@@ -29,7 +30,7 @@ var (
2930
func main() {
3031
flag.Parse()
3132
// uses the current context in kubeconfig
32-
var namespace = "default"
33+
var namespace = "demo"
3334
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
3435
if err != nil {
3536
panic(err.Error())
@@ -41,9 +42,9 @@ func main() {
4142
}
4243

4344
//get the deployment
44-
depName := "janky"
45+
depName := "eggs"
4546
var deployment *v1beta1.Deployment
46-
deployment, err = clientset.Deployments(namespace).Get(depName)
47+
deployment, err = clientset.ExtensionsV1beta1().Deployments(namespace).Get(depName, meta_v1.GetOptions{})
4748
if err != nil {
4849
panic(err.Error())
4950
} else {
@@ -67,7 +68,7 @@ func main() {
6768

6869
//update the deployment labels
6970
newLabels := make(map[string]string)
70-
newLabels["policytest2"] = "pgpolicy"
71+
newLabels["policytest2"] = "jeffsays"
7172

7273
for key, value := range newLabels {
7374
objLabels[key] = value
@@ -97,7 +98,7 @@ func main() {
9798
fmt.Println("created merge patch")
9899
}
99100

100-
_, err = clientset.Deployments(namespace).Patch(depName, api.MergePatchType, patchBytes, "")
101+
_, err = clientset.ExtensionsV1beta1().Deployments(namespace).Patch(depName, types.MergePatchType, patchBytes, "")
101102
if err != nil {
102103
panic("error patching deployment " + err.Error())
103104
}

‎golang-examples/addlabeltopod.go‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
package main
2+
3+
import (
4+
"flag"
5+
"fmt"
6+
jsonpatch "github.com/evanphx/json-patch"
7+
meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
8+
9+
"k8s.io/client-go/kubernetes"
10+
//"k8s.io/client-go/pkg/api"
11+
"k8s.io/apimachinery/pkg/types"
12+
//"k8s.io/client-go/pkg/api/errors"
13+
"encoding/json"
14+
//"k8s.io/api/extensions/v1beta1"
15+
"k8s.io/api/core/v1"
16+
17+
//"k8s.io/client-go/pkg/runtime"
18+
//"k8s.io/client-go/pkg/runtime/serializer"
19+
20+
//"k8s.io/client-go/pkg/api/unversioned"
21+
//"k8s.io/client-go/pkg/api/v1"
22+
//"k8s.io/client-go/rest"
23+
"k8s.io/apimachinery/pkg/api/meta"
24+
25+
"k8s.io/client-go/tools/clientcmd"
26+
)
27+
28+
var (
29+
kubeconfig = flag.String("kubeconfig", "./config", "absolute path to the kubeconfig file")
30+
)
31+
32+
func main() {
33+
flag.Parse()
34+
// uses the current context in kubeconfig
35+
var namespace = "demo"
36+
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
37+
if err != nil {
38+
panic(err.Error())
39+
}
40+
// creates the clientset
41+
clientset, err := kubernetes.NewForConfig(config)
42+
if err != nil {
43+
panic(err.Error())
44+
}
45+
46+
//get the pod
47+
podName := "eggs-d84778bfb-b7669"
48+
var pod *v1.Pod
49+
pod, err = clientset.CoreV1().Pods(namespace).Get(podName, meta_v1.GetOptions{})
50+
if err != nil {
51+
panic(err.Error())
52+
} else {
53+
fmt.Println("got the pod" + pod.Name)
54+
}
55+
origData, err5 := json.Marshal(pod)
56+
if err != nil {
57+
panic(err5)
58+
}
59+
60+
accessor, err2 := meta.Accessor(pod)
61+
if err2 != nil {
62+
panic(err2.Error())
63+
}
64+
65+
objLabels := accessor.GetLabels()
66+
if objLabels == nil {
67+
objLabels = make(map[string]string)
68+
}
69+
fmt.Printf("current labels are %v\n", objLabels)
70+
71+
//update the pod labels
72+
newLabels := make(map[string]string)
73+
newLabels["policytest2"] = "jeffsays"
74+
75+
for key, value := range newLabels {
76+
objLabels[key] = value
77+
}
78+
fmt.Printf("updated labels are %v\n", objLabels)
79+
80+
accessor.SetLabels(objLabels)
81+
//accessor.SetResourceVersion("718408")
82+
83+
newData, err4 := json.Marshal(pod)
84+
if err != nil {
85+
panic(err4)
86+
}
87+
88+
patchBytes, err6 := jsonpatch.CreateMergePatch(origData, newData)
89+
createdPatch := err6 == nil
90+
if err6 != nil {
91+
panic(err6.Error())
92+
}
93+
if len(patchBytes) > 0 {
94+
}
95+
if createdPatch {
96+
fmt.Println("created merge patch")
97+
}
98+
99+
_, err = clientset.CoreV1().Pods(namespace).Patch(podName, types.MergePatchType, patchBytes, "")
100+
if err != nil {
101+
panic("error patching pod " + err.Error())
102+
}
103+
104+
}

‎golang-examples/crdread.go‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package main
2+
3+
import (
4+
"flag"
5+
log "github.com/Sirupsen/logrus"
6+
crv1 "github.com/crunchydata/postgres-operator/apis/cr/v1"
7+
clientset "github.com/crunchydata/postgres-operator/client"
8+
"k8s.io/client-go/kubernetes"
9+
"k8s.io/client-go/tools/clientcmd"
10+
)
11+
12+
var (
13+
kubeconfig = flag.String("kubeconfig", "./config", "absolute path to the kubeconfig file")
14+
)
15+
16+
func main() {
17+
flag.Parse()
18+
// uses the current context in kubeconfig
19+
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
20+
if err != nil {
21+
panic(err.Error())
22+
}
23+
24+
kubeClient, err2 := kubernetes.NewForConfig(config)
25+
if err2 != nil {
26+
panic(err2.Error())
27+
}
28+
if kubeClient != nil {
29+
log.Println("got kube client")
30+
}
31+
32+
restclient, _, err := clientset.NewClient(config)
33+
if err != nil {
34+
panic(err)
35+
}
36+
log.Println("got rest client")
37+
38+
taskName := "goober" + "-failover"
39+
//get the task
40+
task := crv1.Pgtask{}
41+
err = restclient.Get().
42+
Resource(crv1.PgtaskResourcePlural).
43+
Namespace("demo").
44+
Name(taskName).
45+
Do().
46+
Into(&task)
47+
if err != nil {
48+
log.Error("error getting pgtask " + taskName)
49+
log.Error(err)
50+
return
51+
}
52+
53+
log.Println("got pgtask " + task.Name)
54+
55+
}

‎operator/cluster/cluster.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ import (
3333
type Strategy interface {
3434
Scale(*kubernetes.Clientset, *rest.RESTClient, *crv1.Pgreplica, string, string, *crv1.Pgcluster) error
3535
AddCluster(*kubernetes.Clientset, *rest.RESTClient, *crv1.Pgcluster, string, string) error
36-
Failover(*kubernetes.Clientset, *rest.RESTClient, string, string, string) error
36+
Failover(*kubernetes.Clientset, *rest.RESTClient, string, *crv1.Pgtask, string, *rest.Config) error
3737
CreateReplica(string, *kubernetes.Clientset, *crv1.Pgcluster, string, string, string) error
3838
DeleteCluster(*kubernetes.Clientset, *rest.RESTClient, *crv1.Pgcluster, string) error
3939

‎operator/cluster/cluster_strategy_1.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -348,6 +348,7 @@ func (r Strategy1) CreateReplica(serviceName string, clientset *kubernetes.Clien
348348
clusterName := cl.Spec.ClusterName
349349

350350
replicaLabels := getPrimaryLabels(serviceName, clusterName, true, cl.Spec.UserLabels)
351+
351352
//create the replica deployment
352353
replicaDeploymentFields := DeploymentTemplateFields{
353354
Name: depName,
@@ -561,6 +562,7 @@ func (r Strategy1) Scale(clientset *kubernetes.Clientset, client *rest.RESTClien
561562
replicaFlag := true
562563

563564
replicaLabels := getPrimaryLabels(serviceName, replica.Spec.ClusterName, replicaFlag, cluster.Spec.UserLabels)
565+
replicaLabels["replica-name"] = replica.Spec.Name
564566

565567
//create the replica deployment
566568
replicaDeploymentFields := DeploymentTemplateFields{

‎operator/cluster/failover.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import (
2626
)
2727

2828
// FailoverBase ...
29-
func FailoverBase(namespace string, clientset *kubernetes.Clientset, client *rest.RESTClient, task *crv1.Pgtask) {
29+
func FailoverBase(namespace string, clientset *kubernetes.Clientset, client *rest.RESTClient, task *crv1.Pgtask, restconfig *rest.Config) {
3030
var err error
3131

3232
//look up the pgcluster for this task
@@ -58,6 +58,6 @@ func FailoverBase(namespace string, clientset *kubernetes.Clientset, client *res
5858
return
5959
}
6060

61-
strategy.Failover(clientset, client, clusterName, task.ObjectMeta.Labels["target"], namespace)
61+
strategy.Failover(clientset, client, clusterName, task, namespace, restconfig)
6262

6363
}

0 commit comments

Comments
 (0)