...
1
2
3
4
5
6
7
8
9
10
11
12
13
14 package podnetworkchaosmanager
15
16 import (
17 "context"
18 "errors"
19
20 "github.com/go-logr/logr"
21 v1 "k8s.io/api/core/v1"
22 k8sError "k8s.io/apimachinery/pkg/api/errors"
23 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
24 "k8s.io/apimachinery/pkg/runtime"
25 "k8s.io/apimachinery/pkg/types"
26 "k8s.io/client-go/util/retry"
27 "sigs.k8s.io/controller-runtime/pkg/client"
28
29 "github.com/chaos-mesh/chaos-mesh/api/v1alpha1"
30 )
31
32 var (
33
34 ErrPodNotFound = errors.New("pod not found")
35
36
37
38 ErrPodNotRunning = errors.New("pod not running")
39 )
40
41
42 type PodNetworkManager struct {
43 Source string
44
45 Log logr.Logger
46 client.Client
47 client.Reader
48 scheme *runtime.Scheme
49
50 Key types.NamespacedName
51 T *PodNetworkTransaction
52 }
53
54
55 type CommitResponse struct {
56 Key types.NamespacedName
57 Err error
58 }
59
60
61 func (m *PodNetworkManager) Commit(ctx context.Context, owner *v1alpha1.NetworkChaos) (int64, error) {
62 m.Log.Info("running modification on pod", "key", m.Key, "modification", m.T)
63 updateError := retry.RetryOnConflict(retry.DefaultRetry, func() error {
64 chaos := &v1alpha1.PodNetworkChaos{}
65
66 err := m.Client.Get(ctx, m.Key, chaos)
67 if err != nil {
68 if !k8sError.IsNotFound(err) {
69 m.Log.Error(err, "error while getting podnetworkchaos")
70 return err
71 }
72
73 err := m.CreateNewPodNetworkChaos(ctx)
74 if err != nil {
75 m.Log.Error(err, "error while creating new podnetworkchaos")
76 return err
77 }
78
79 return nil
80 }
81
82 err = m.T.Apply(chaos)
83 if err != nil {
84 m.Log.Error(err, "error while applying transactions", "transaction", m.T)
85 return err
86 }
87
88 return m.Client.Update(ctx, chaos)
89 })
90 if updateError != nil {
91 return 0, updateError
92 }
93
94 chaos := &v1alpha1.PodNetworkChaos{}
95 err := m.Reader.Get(ctx, m.Key, chaos)
96 if err != nil {
97 m.Log.Error(err, "error while getting the latest generation number")
98 return 0, err
99 }
100 return chaos.GetGeneration(), nil
101 }
102
103 func (m *PodNetworkManager) CreateNewPodNetworkChaos(ctx context.Context) error {
104 var err error
105 chaos := &v1alpha1.PodNetworkChaos{}
106
107 pod := v1.Pod{}
108 err = m.Client.Get(ctx, m.Key, &pod)
109 if err != nil {
110 if !k8sError.IsNotFound(err) {
111 m.Log.Error(err, "error while finding pod")
112 return err
113 }
114
115 m.Log.Info("pod not found", "key", m.Key, "error", err.Error())
116 err = ErrPodNotFound
117 return err
118 }
119
120 if pod.Status.Phase != v1.PodRunning {
121 m.Log.Info("pod is not running", "key", m.Key)
122 err = ErrPodNotRunning
123 return err
124 }
125
126 chaos.Name = m.Key.Name
127 chaos.Namespace = m.Key.Namespace
128 chaos.OwnerReferences = []metav1.OwnerReference{
129 {
130 APIVersion: pod.APIVersion,
131 Kind: pod.Kind,
132 Name: pod.Name,
133 UID: pod.UID,
134 },
135 }
136 err = m.T.Apply(chaos)
137 if err != nil {
138 m.Log.Error(err, "error while applying transactions", "transaction", m.T)
139 return err
140 }
141
142 return m.Client.Create(ctx, chaos)
143 }
144