11package firehose
22
33import (
4- "context"
54 _ "embed"
6- "encoding/json"
75
86 "github.com/odpf/entropy/core/module"
9- "github.com/odpf/entropy/core/resource"
107 "github.com/odpf/entropy/modules/kubernetes"
11- "github.com/odpf/entropy/pkg/errors"
12- "github.com/odpf/entropy/pkg/helm"
13- "github.com/odpf/entropy/pkg/kube"
148)
159
1610const (
@@ -20,15 +14,16 @@ const (
2014)
2115
2216const (
23- helmCreate = "helm_create"
24- helmUpdate = "helm_update"
17+ releaseCreate = "release_create"
18+ releaseUpdate = "release_update"
19+ )
2520
21+ const (
2622 stateRunning = "RUNNING"
2723 stateStopped = "STOPPED"
2824)
2925
3026const (
31- keyReplicaCount = "replicaCount"
3227 keyKubeDependency = "kube_cluster"
3328)
3429
@@ -66,176 +61,3 @@ var Module = module.Descriptor{
6661}
6762
6863type firehoseModule struct {}
69-
70- func (m * firehoseModule ) Plan (_ context.Context , spec module.Spec , act module.ActionRequest ) (* resource.Resource , error ) {
71- if act .Name == module .CreateAction {
72- return m .planCreate (spec , act )
73- }
74- return m .planChange (spec , act )
75- }
76-
77- func (m * firehoseModule ) Sync (_ context.Context , spec module.Spec ) (* resource.State , error ) {
78- r := spec .Resource
79-
80- var data moduleData
81- if err := json .Unmarshal (r .State .ModuleData , & data ); err != nil {
82- return nil , err
83- }
84-
85- if len (data .PendingSteps ) == 0 {
86- return & resource.State {
87- Status : resource .StatusCompleted ,
88- Output : r .State .Output ,
89- ModuleData : r .State .ModuleData ,
90- }, nil
91- }
92-
93- pendingStep := data .PendingSteps [0 ]
94- data .PendingSteps = data .PendingSteps [1 :]
95-
96- var conf moduleConfig
97- if err := json .Unmarshal (r .Spec .Configs , & conf ); err != nil {
98- return nil , errors .ErrInvalid .WithMsgf ("invalid config json: %v" , err )
99- }
100-
101- var kubeOut kubernetes.Output
102- if err := json .Unmarshal (spec .Dependencies [keyKubeDependency ].Output , & kubeOut ); err != nil {
103- return nil , err
104- }
105-
106- if err := m .helmSync (pendingStep == helmCreate , conf , kubeOut ); err != nil {
107- return nil , err
108- }
109-
110- return & resource.State {
111- Status : resource .StatusCompleted ,
112- Output : Output {
113- Namespace : conf .ReleaseConfigs .Namespace ,
114- ReleaseName : conf .ReleaseConfigs .Name ,
115- }.JSON (),
116- ModuleData : data .JSON (),
117- }, nil
118- }
119-
120- func (* firehoseModule ) Log (ctx context.Context , spec module.Spec , filter map [string ]string ) (<- chan module.LogChunk , error ) {
121- r := spec .Resource
122-
123- var conf moduleConfig
124- if err := json .Unmarshal (r .Spec .Configs , & conf ); err != nil {
125- return nil , errors .ErrInvalid .WithMsgf ("invalid config json: %v" , err )
126- }
127-
128- var kubeOut kubernetes.Output
129- if err := json .Unmarshal (spec .Dependencies [keyKubeDependency ].Output , & kubeOut ); err != nil {
130- return nil , err
131- }
132-
133- if filter == nil {
134- filter = make (map [string ]string )
135- }
136- filter ["app" ] = conf .ReleaseConfigs .Name
137-
138- kubeCl := kube .NewClient (kubeOut .Configs )
139- logs , err := kubeCl .StreamLogs (ctx , defaultNamespace , filter )
140- if err != nil {
141- return nil , err
142- }
143-
144- mappedLogs := make (chan module.LogChunk )
145- go func () {
146- defer close (mappedLogs )
147- for {
148- select {
149- case log , ok := <- logs :
150- if ! ok {
151- return
152- }
153- mappedLogs <- module.LogChunk {Data : log .Data , Labels : log .Labels }
154- case <- ctx .Done ():
155- return
156- }
157- }
158- }()
159-
160- return mappedLogs , err
161- }
162-
163- func (* firehoseModule ) planCreate (spec module.Spec , act module.ActionRequest ) (* resource.Resource , error ) {
164- r := spec .Resource
165-
166- var reqConf moduleConfig
167- if err := json .Unmarshal (act .Params , & reqConf ); err != nil {
168- return nil , errors .ErrInvalid .WithMsgf ("invalid config json: %v" , err )
169- } else {
170- reqConf .sanitiseAndValidate (r )
171- }
172-
173- r .Spec .Configs = reqConf .JSON ()
174- r .State = resource.State {
175- Status : resource .StatusPending ,
176- ModuleData : moduleData {
177- PendingSteps : []string {helmCreate },
178- }.JSON (),
179- }
180- return & r , nil
181- }
182-
183- func (* firehoseModule ) planChange (spec module.Spec , act module.ActionRequest ) (* resource.Resource , error ) {
184- r := spec .Resource
185-
186- var conf moduleConfig
187- if err := json .Unmarshal (r .Spec .Configs , & conf ); err != nil {
188- return nil , errors .ErrInvalid .WithMsgf ("invalid config json: %v" , err )
189- }
190-
191- switch act .Name {
192- case module .UpdateAction :
193- var reqConf moduleConfig
194- if err := json .Unmarshal (act .Params , & reqConf ); err != nil {
195- return nil , errors .ErrInvalid .WithMsgf ("invalid config json: %v" , err )
196- }
197- conf = reqConf
198-
199- case ScaleAction :
200- var scaleParams struct {
201- Replicas int `json:"replicas"`
202- }
203- if err := json .Unmarshal (act .Params , & scaleParams ); err != nil {
204- return nil , errors .ErrInvalid .WithMsgf ("invalid config json: %v" , err )
205- }
206- conf .ReleaseConfigs .Values [keyReplicaCount ] = scaleParams .Replicas
207-
208- case StartAction :
209- conf .State = stateRunning
210-
211- case StopAction :
212- conf .State = stateStopped
213- }
214-
215- conf .sanitiseAndValidate (r )
216- r .Spec .Configs = conf .JSON ()
217- r .State = resource.State {
218- Status : resource .StatusPending ,
219- ModuleData : moduleData {
220- PendingSteps : []string {helmUpdate },
221- }.JSON (),
222- }
223- return & r , nil
224- }
225-
226- func (* firehoseModule ) helmSync (isCreate bool , conf moduleConfig , kube kubernetes.Output ) error {
227- helmCl := helm .NewClient (& helm.Config {Kubernetes : kube .Configs })
228-
229- if conf .State == stateStopped {
230- conf .ReleaseConfigs .Values [keyReplicaCount ] = 0
231- }
232-
233- var helmErr error
234- if isCreate {
235- _ , helmErr = helmCl .Create (& conf .ReleaseConfigs )
236- } else {
237- _ , helmErr = helmCl .Update (& conf .ReleaseConfigs )
238- }
239-
240- return helmErr
241- }
0 commit comments