feat: Log operations for statistics clear

keboola · Jan 23, 2025 · e89175c · e89175c
1 parent f1e7a21
commit e89175c
Show file tree

Hide file tree

Showing 7 changed files with 35 additions and 7 deletions.
diff --git a/internal/pkg/service/common/etcdop/op/atomic_do.go b/internal/pkg/service/common/etcdop/op/atomic_do.go
@@ -16,12 +16,14 @@ func (v *AtomicOp[R]) Do(ctx context.Context, opts ...Option) AtomicResult[R] {
 	attempt := 0
 
 	var ok bool
+	var ops int
 	var err error
 	var header *Header
 
 	for {
 		txnResult := v.DoWithoutRetry(ctx, opts...)
 		ok = txnResult.Succeeded()
+		ops = txnResult.TotalOps()
 		err = txnResult.Err()
 		header = txnResult.Header()
 		if err == nil && !ok {
@@ -44,7 +46,7 @@ func (v *AtomicOp[R]) Do(ctx context.Context, opts ...Option) AtomicResult[R] {
 		)
 	}
 
-	return AtomicResult[R]{result: v.result, error: err, header: header, attempt: attempt, elapsedTime: elapsedTime}
+	return AtomicResult[R]{result: v.result, ops: ops, error: err, header: header, attempt: attempt, elapsedTime: elapsedTime}
 }
 
 func (v *AtomicOp[R]) DoWithoutRetry(ctx context.Context, opts ...Option) *TxnResult[R] {

diff --git a/internal/pkg/service/common/etcdop/op/atomic_result.go b/internal/pkg/service/common/etcdop/op/atomic_result.go
@@ -6,6 +6,7 @@ import (
 
 type AtomicResult[R any] struct {
 	result      *R
+	ops         int
 	error       error
 	header      *Header
 	attempt     int
@@ -20,6 +21,10 @@ func (v AtomicResult[R]) Result() R {
 	return *v.result
 }
 
+func (v AtomicResult[R]) TotalOps() int {
+	return v.ops
+}
+
 func (v AtomicResult[R]) Err() error {
 	return v.error
 }

diff --git a/internal/pkg/service/common/etcdop/op/txn.go b/internal/pkg/service/common/etcdop/op/txn.go
@@ -219,7 +219,9 @@ func (v *TxnOp[R]) OnSucceeded(fn func(result *TxnResult[R])) *TxnOp[R] {
 
 func (v *TxnOp[R]) Do(ctx context.Context, opts ...Option) *TxnResult[R] {
 	if lowLevel, err := v.lowLevelTxn(ctx); err == nil {
-		return lowLevel.Do(ctx, opts...)
+		result := lowLevel.Do(ctx, opts...)
+		result.addTotals(len(lowLevel.ifs), len(lowLevel.thenOps), len(lowLevel.elseOps))
+		return result
 	} else {
 		return newErrorTxnResult[R](err)
 	}

diff --git a/internal/pkg/service/common/etcdop/op/txn_result.go b/internal/pkg/service/common/etcdop/op/txn_result.go
@@ -3,7 +3,10 @@ package op
 // TxnResult is result of the TxnOp.
 type TxnResult[R any] struct {
 	*resultBase
-	result *R
+	result       *R
+	totalIfs     int
+	totalThenOps int
+	totalElseOps int
 }
 
 func newTxnResult[R any](base *resultBase, result *R) *TxnResult[R] {
@@ -16,6 +19,16 @@ func newErrorTxnResult[R any](err error) *TxnResult[R] {
 	return r
 }
 
+func (v *TxnResult[R]) addTotals(totalIfs, totalThenOps, totalElseOps int) {
+	v.totalIfs = totalIfs
+	v.totalThenOps = totalThenOps
+	v.totalElseOps = totalElseOps
+}
+
+func (v *TxnResult[R]) TotalOps() int {
+	return v.totalIfs + v.totalThenOps + v.totalElseOps
+}
+
 func (v *TxnResult[R]) Succeeded() bool {
 	return v.response != nil && v.response.Txn() != nil && v.response.Txn().Succeeded
 }

diff --git a/internal/pkg/service/stream/api/service/sink.go b/internal/pkg/service/stream/api/service/sink.go
@@ -364,7 +364,10 @@ func (s *service) SinkStatisticsClear(ctx context.Context, d dependencies.SinkRe
 		return err
 	}
 
-	return d.StatisticsRepository().ResetSinkStats(d.SinkKey()).Do(ctx).Err()
+	result := d.StatisticsRepository().ResetSinkStats(d.SinkKey()).Do(ctx)
+	s.logger.Infof(ctx, `Statistics clear for sink "%s" used %d operations`, d.SinkKey().String(), result.TotalOps())
+
+	return result.Err()
 }
 
 func (s *service) DisableSink(ctx context.Context, d dependencies.SinkRequestScope, payload *api.DisableSinkPayload) (res *api.Task, err error) {

diff --git a/internal/pkg/service/stream/api/service/source.go b/internal/pkg/service/stream/api/service/source.go
@@ -343,7 +343,10 @@ func (s *service) SourceStatisticsClear(ctx context.Context, d dependencies.Sour
 		sinkKeys = append(sinkKeys, sink.SinkKey)
 	}
 
-	return d.StatisticsRepository().ResetAllSinksStats(ctx, sinkKeys)
+	result := d.StatisticsRepository().ResetAllSinksStats(sinkKeys).Do(ctx)
+	s.logger.Infof(ctx, `Statistics clear for source "%s" used %d operations`, d.SourceKey().String(), result.TotalOps())
+
+	return result.Err()
 }
 
 func (s *service) DisableSource(ctx context.Context, d dependencies.SourceRequestScope, payload *api.DisableSourcePayload) (*api.Task, error) {

diff --git a/internal/pkg/service/stream/storage/statistics/repository/stats_reset.go b/internal/pkg/service/stream/storage/statistics/repository/stats_reset.go
@@ -12,14 +12,14 @@ import (
 	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/statistics"
 )
 
-func (r *Repository) ResetAllSinksStats(ctx context.Context, sinkKeys []key.SinkKey) error {
+func (r *Repository) ResetAllSinksStats(sinkKeys []key.SinkKey) *op.AtomicOp[op.NoResult] {
 	ops := op.Atomic(r.client, &op.NoResult{})
 
 	for _, sinkKey := range sinkKeys {
 		ops.AddFrom(r.ResetSinkStats(sinkKey))
 	}
 
-	return ops.Do(ctx).Err()
+	return ops
 }
 
 // ResetSinkStats sums all statistics for data in target level and saves the sum in _reset key.