@@ -142,11 +142,13 @@ func (d *dividerWorker) StopProcessing(ctx context.Context, works ...string) err
142142 start := time .Now ()
143143 err := d .storage .RemoveWorkFromDividedWork (ctx , works )
144144 if err != nil {
145+ ObserveInc (RemoveWorkFromDividerError , d .conf .metricsName , 1 )
145146 return errors .Wrap (err , "failed to Remove Work From Divided Work" )
146147 }
147148 for _ , work := range works {
148149 err = d .removeWork .Publish (ctx , work )
149150 if err != nil {
151+ ObserveInc (PublishRemoveWorkFromDividerError , d .conf .metricsName , 1 )
150152 return errors .Wrap (err , "failed to publish work removal" )
151153 }
152154 }
@@ -164,13 +166,15 @@ func (d *dividerWorker) StartProcessing(ctx context.Context, works ...string) er
164166
165167 err := d .storage .AddWorkToDividedWork (ctx , works )
166168 if err != nil {
169+ ObserveInc (AddWorkToDividerError , d .conf .metricsName , 1 )
167170 return errors .Wrap (err , "failed to Add Work To Divided Work" )
168171 }
169172 for _ , work := range works {
170173 d .conf .logger (ctx ).Debug ("sending start processing for: " + work , slog .String ("divider.id" , d .conf .instanceID ))
171174
172175 err = d .newWork .Publish (ctx , work )
173176 if err != nil {
177+ ObserveInc (PublishAddWorkToDividerError , d .conf .metricsName , 1 )
174178 return errors .Wrap (err , "failed to publish work start" )
175179 }
176180 }
@@ -198,6 +202,7 @@ func (d *dividerWorker) newWorkerEvent(ctx context.Context, key string) {
198202 d .conf .logger (ctx ).Debug ("newWorkerEvent triggered: " + key , slog .String ("divider.id" , d .conf .instanceID ))
199203 err := d .rectifyWork (ctx )
200204 if err != nil {
205+ ObserveInc (RectifyWorkError , d .conf .metricsName , 1 )
201206 d .conf .logger (ctx ).Error ("failed to rectify work" , slog .String ("err.error" , err .Error ()), slog .String ("divider.id" , d .conf .instanceID ))
202207 }
203208
@@ -219,6 +224,7 @@ func (d *dividerWorker) removeWorkerEvent(ctx context.Context, key string) {
219224 //rectify all of your work.
220225 err := d .rectifyWork (ctx )
221226 if err != nil {
227+ ObserveInc (RectifyWorkError , d .conf .metricsName , 1 )
222228 d .conf .logger (ctx ).Error ("failed to rectify work" , slog .String ("err.error" , err .Error ()), slog .String ("divider.id" , d .conf .instanceID ))
223229 }
224230
@@ -236,12 +242,14 @@ func (d *dividerWorker) newWorkEvent(ctx context.Context, key string) {
236242 for _ , v := range d .getWorkerNodeKeys () {
237243 inRange , err := d .storage .CheckWorkInKeyRange (ctx , v , key )
238244 if err != nil {
245+ ObserveInc (CheckWorkInKeyRangeError , d .conf .metricsName , 1 )
239246 d .conf .logger (ctx ).Error ("failed to check if work is in the range of this worker" , slog .String ("err.error" , err .Error ()), slog .String ("divider.id" , d .conf .instanceID ))
240247 return
241248 }
242249 if inRange {
243250 err = d .conf .starter (ctx , key )
244251 if err != nil {
252+ ObserveInc (StartWorkExternalError , d .conf .metricsName , 1 )
245253 d .conf .logger (ctx ).Error ("failed to execute starter" , slog .String ("err.error" , err .Error ()), slog .String ("divider.id" , d .conf .instanceID ))
246254 return
247255 }
@@ -480,6 +488,8 @@ func (d *dividerWorker) rectifyWork(ctx context.Context) (err error) {
480488 defer can ()
481489 err = d .conf .stopper (ctx2 , key )
482490 if err != nil {
491+ ObserveInc (StopWorkExternalError , d .conf .metricsName , 1 )
492+
483493 d .conf .logger (ctx2 ).Error ("failed to execute stopper, not removing from known work" , slog .String ("err.error" , err .Error ()), slog .String ("divider.id" , d .conf .instanceID ))
484494 continue
485495 }
@@ -492,6 +502,7 @@ func (d *dividerWorker) rectifyWork(ctx context.Context) (err error) {
492502 defer can ()
493503 err = d .conf .starter (ctx2 , key )
494504 if err != nil {
505+ ObserveInc (StartWorkExternalError , d .conf .metricsName , 1 )
495506 d .conf .logger (ctx2 ).Error ("failed to execute starter, not adding to known work" , slog .String ("err.error" , err .Error ()), slog .String ("divider.id" , d .conf .instanceID ))
496507 continue
497508 }
0 commit comments