@@ -22,23 +22,21 @@ const (
2222// WAL for calcium.
2323type WAL struct {
2424 wal.WAL
25- config types.Config
2625 calcium * Calcium
2726}
2827
29- func newCalciumWAL (cal * Calcium ) (* WAL , error ) {
30- w := & WAL {
31- WAL : wal .NewHydro (),
32- config : cal .config ,
33- calcium : cal ,
28+ func newWAL (config types.Config , calcium * Calcium ) (* WAL , error ) {
29+ hydro , err := wal .NewHydro (config .WALFile , config .WALOpenTimeout )
30+ if err != nil {
31+ return nil , err
3432 }
3533
36- if err := w .WAL .Open (w .config .WALFile , w .config .WALOpenTimeout ); err != nil {
37- return nil , err
34+ w := & WAL {
35+ WAL : hydro ,
36+ calcium : calcium ,
3837 }
3938
4039 w .registerHandlers ()
41-
4240 return w , nil
4341}
4442
@@ -49,105 +47,22 @@ func (w *WAL) registerHandlers() {
4947 w .Register (newProcessingCreatedHandler (w .calcium ))
5048}
5149
52- func (w * WAL ) logCreateLambda (opts * types.CreateWorkloadMessage ) (wal.Commit , error ) {
53- return w .Log (eventCreateLambda , opts .WorkloadID )
54- }
55-
56- // CreateWorkloadHandler indicates event handler for creating workload.
57- type CreateWorkloadHandler struct {
58- event string
59- calcium * Calcium
60- }
61-
62- func newCreateWorkloadHandler (cal * Calcium ) * CreateWorkloadHandler {
63- return & CreateWorkloadHandler {
64- event : eventWorkloadCreated ,
65- calcium : cal ,
66- }
67- }
68-
69- // Event .
70- func (h * CreateWorkloadHandler ) Event () string {
71- return h .event
72- }
73-
74- // Check .
75- func (h * CreateWorkloadHandler ) Check (ctx context.Context , raw interface {}) (handle bool , err error ) {
76- _ , ok := raw .(* types.Workload )
77- if ! ok {
78- return false , types .NewDetailedErr (types .ErrInvalidType , raw )
79- }
80- return true , nil
81- }
82-
83- // Encode .
84- func (h * CreateWorkloadHandler ) Encode (raw interface {}) ([]byte , error ) {
85- wrk , ok := raw .(* types.Workload )
86- if ! ok {
87- return nil , types .NewDetailedErr (types .ErrInvalidType , raw )
88- }
89- return json .Marshal (wrk )
90- }
91-
92- // Decode .
93- func (h * CreateWorkloadHandler ) Decode (bs []byte ) (interface {}, error ) {
94- wrk := & types.Workload {}
95- err := json .Unmarshal (bs , wrk )
96- return wrk , err
97- }
98-
99- // Handle will remove instance, remove meta, restore resource
100- func (h * CreateWorkloadHandler ) Handle (ctx context.Context , raw interface {}) (err error ) {
101- wrk , _ := raw .(* types.Workload )
102- logger := log .WithField ("WAL.Handle" , "CreateWorkload" ).WithField ("ID" , wrk .ID ).WithField ("nodename" , wrk .Nodename )
103-
104- ctx , cancel := getReplayContext (ctx )
105- defer cancel ()
106-
107- if _ , err = h .calcium .GetWorkload (ctx , wrk .ID ); err == nil {
108- // workload meta exists
109- ch , err := h .calcium .RemoveWorkload (ctx , []string {wrk .ID }, true , 0 )
110- if err != nil {
111- return logger .Err (ctx , err )
112- }
113- for msg := range ch {
114- if ! msg .Success {
115- logger .Errorf (ctx , "failed to remove workload" )
116- }
117- }
118- logger .Infof (ctx , "workload with meta removed" )
119- return nil
120- }
121-
122- // workload meta doesn't exist
123- node , err := h .calcium .GetNode (ctx , wrk .Nodename )
124- if err != nil {
125- return logger .Err (ctx , err )
126- }
127- if err = node .Engine .VirtualizationRemove (ctx , wrk .ID , true , true ); err != nil && ! errors .Is (err , types .ErrWorkloadNotExists ) {
128- return logger .Err (ctx , err )
129- }
130-
131- logger .Infof (ctx , "workload removed" )
132- return nil
133- }
134-
13550// CreateLambdaHandler indicates event handler for creating lambda.
13651type CreateLambdaHandler struct {
137- event string
52+ typ string
13853 calcium * Calcium
13954}
14055
141- func newCreateLambdaHandler (cal * Calcium ) * CreateLambdaHandler {
56+ func newCreateLambdaHandler (calcium * Calcium ) * CreateLambdaHandler {
14257 return & CreateLambdaHandler {
143- event : eventCreateLambda ,
144- calcium : cal ,
58+ typ : eventCreateLambda ,
59+ calcium : calcium ,
14560 }
14661}
14762
14863// Event .
149- func (h * CreateLambdaHandler ) Event () string {
150- return h .event
64+ func (h * CreateLambdaHandler ) Typ () string {
65+ return h .typ
15166}
15267
15368// Check .
@@ -202,26 +117,90 @@ func (h *CreateLambdaHandler) Handle(ctx context.Context, raw interface{}) error
202117 return nil
203118}
204119
205- func getReplayContext (ctx context.Context ) (context.Context , context.CancelFunc ) {
206- return context .WithTimeout (ctx , time .Second * 32 )
120+ // CreateWorkloadHandler indicates event handler for creating workload.
121+ type CreateWorkloadHandler struct {
122+ typ string
123+ calcium * Calcium
124+ }
125+
126+ func newCreateWorkloadHandler (calcium * Calcium ) * CreateWorkloadHandler {
127+ return & CreateWorkloadHandler {
128+ typ : eventWorkloadCreated ,
129+ calcium : calcium ,
130+ }
131+ }
132+
133+ // Event .
134+ func (h * CreateWorkloadHandler ) Typ () string {
135+ return h .typ
136+ }
137+
138+ // Check .
139+ func (h * CreateWorkloadHandler ) Check (ctx context.Context , raw interface {}) (handle bool , err error ) {
140+ _ , ok := raw .(* types.Workload )
141+ if ! ok {
142+ return false , types .NewDetailedErr (types .ErrInvalidType , raw )
143+ }
144+ return true , nil
145+ }
146+
147+ // Encode .
148+ func (h * CreateWorkloadHandler ) Encode (raw interface {}) ([]byte , error ) {
149+ wrk , ok := raw .(* types.Workload )
150+ if ! ok {
151+ return nil , types .NewDetailedErr (types .ErrInvalidType , raw )
152+ }
153+ return json .Marshal (wrk )
154+ }
155+
156+ // Decode .
157+ func (h * CreateWorkloadHandler ) Decode (bs []byte ) (interface {}, error ) {
158+ wrk := & types.Workload {}
159+ err := json .Unmarshal (bs , wrk )
160+ return wrk , err
161+ }
162+
163+ // Handle will remove instance, remove meta, restore resource
164+ func (h * CreateWorkloadHandler ) Handle (ctx context.Context , raw interface {}) (err error ) {
165+ wrk , _ := raw .(* types.Workload )
166+ logger := log .WithField ("WAL.Handle" , "CreateWorkload" ).WithField ("ID" , wrk .ID ).WithField ("nodename" , wrk .Nodename )
167+
168+ ctx , cancel := getReplayContext (ctx )
169+ defer cancel ()
170+
171+ if _ , err = h .calcium .GetWorkload (ctx , wrk .ID ); err == nil {
172+ return h .calcium .doRemoveWorkloadSync (ctx , []string {wrk .ID })
173+ }
174+
175+ // workload meta doesn't exist
176+ node , err := h .calcium .GetNode (ctx , wrk .Nodename )
177+ if err != nil {
178+ return logger .Err (ctx , err )
179+ }
180+ if err = node .Engine .VirtualizationRemove (ctx , wrk .ID , true , true ); err != nil && ! errors .Is (err , types .ErrWorkloadNotExists ) {
181+ return logger .Err (ctx , err )
182+ }
183+
184+ logger .Infof (ctx , "workload removed" )
185+ return nil
207186}
208187
209188// WorkloadResourceAllocatedHandler .
210189type WorkloadResourceAllocatedHandler struct {
211- event string
190+ typ string
212191 calcium * Calcium
213192}
214193
215- func newWorkloadResourceAllocatedHandler (cal * Calcium ) * WorkloadResourceAllocatedHandler {
194+ func newWorkloadResourceAllocatedHandler (calcium * Calcium ) * WorkloadResourceAllocatedHandler {
216195 return & WorkloadResourceAllocatedHandler {
217- event : eventWorkloadResourceAllocated ,
218- calcium : cal ,
196+ typ : eventWorkloadResourceAllocated ,
197+ calcium : calcium ,
219198 }
220199}
221200
222201// Event .
223- func (h * WorkloadResourceAllocatedHandler ) Event () string {
224- return h .event
202+ func (h * WorkloadResourceAllocatedHandler ) Typ () string {
203+ return h .typ
225204}
226205
227206// Check .
@@ -276,20 +255,20 @@ func (h *WorkloadResourceAllocatedHandler) Handle(ctx context.Context, raw inter
276255
277256// ProcessingCreatedHandler .
278257type ProcessingCreatedHandler struct {
279- event string
258+ typ string
280259 calcium * Calcium
281260}
282261
283- func newProcessingCreatedHandler (cal * Calcium ) * ProcessingCreatedHandler {
262+ func newProcessingCreatedHandler (calcium * Calcium ) * ProcessingCreatedHandler {
284263 return & ProcessingCreatedHandler {
285- event : eventProcessingCreated ,
286- calcium : cal ,
264+ typ : eventProcessingCreated ,
265+ calcium : calcium ,
287266 }
288267}
289268
290269// Event .
291- func (h * ProcessingCreatedHandler ) Event () string {
292- return h .event
270+ func (h * ProcessingCreatedHandler ) Typ () string {
271+ return h .typ
293272}
294273
295274// Check .
@@ -329,3 +308,7 @@ func (h *ProcessingCreatedHandler) Handle(ctx context.Context, raw interface{})
329308 logger .Infof (ctx , "obsolete processing deleted" )
330309 return
331310}
311+
312+ func getReplayContext (ctx context.Context ) (context.Context , context.CancelFunc ) {
313+ return context .WithTimeout (ctx , time .Second * 32 ) // TODO why 32?
314+ }
0 commit comments