nuclio · TomerShor · Apr 6, 2025 · Apr 2, 2025 · Apr 2, 2025 · Apr 2, 2025
diff --git a/pkg/processor/eventprocessor/pool.go b/pkg/processor/eventprocessor/pool.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/nuclio/nuclio/pkg/common"
 	"github.com/nuclio/nuclio/pkg/errgroup"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
@@ -162,16 +163,16 @@ func (sa *syncPoolAllocator) SetObjects(objects []EventProcessor) error {
 // GetStatistics returns object allocator statistics
 // return unsafe copy of the statistics to avoid any unnecessary blocking of the actual statistics object
 // used in gatherers which are thread-safe
-func (sa *syncPoolAllocator) GetStatistics() *AllocatorStatistics {
-	statistics := &AllocatorStatistics{
+func (sa *syncPoolAllocator) GetStatistics() *statistics.AllocatorStatistics {
+	allocatorStatistics := &statistics.AllocatorStatistics{
 		AllocationCount:                       sa.statistics.AllocationCount.Load(),
 		AllocationSuccessImmediateTotal:       sa.statistics.AllocationSuccessImmediateTotal.Load(),
 		AllocationSuccessAfterWaitTotal:       sa.statistics.AllocationSuccessAfterWaitTotal.Load(),
 		AllocationTimeoutTotal:                sa.statistics.AllocationTimeoutTotal.Load(),
 		AllocationWaitDurationMilliSecondsSum: sa.statistics.AllocationWaitDurationMilliSecondsSum.Load(),
 		AllocationObjectsAvailablePercentage:  sa.statistics.AllocationObjectsAvailablePercentage.Load(),
 	}
-	return statistics
+	return allocatorStatistics
 }
 
 func (sa *syncPoolAllocator) SignalDraining() error {

diff --git a/pkg/processor/eventprocessor/singleton.go b/pkg/processor/eventprocessor/singleton.go
@@ -20,6 +20,8 @@ import (
 	"errors"
 	"time"
 
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
+
 	"github.com/nuclio/logger"
 )
 
@@ -64,8 +66,8 @@ func (s *asyncSingletonAllocator) GetNumObjectsAvailable() int {
 }
 
 // GetStatistics returns allocator statistics
-func (s *asyncSingletonAllocator) GetStatistics() *AllocatorStatistics {
-	return nil
+func (s *asyncSingletonAllocator) GetStatistics() *statistics.AllocatorStatistics {
+	return s.object.GetAllocationStatistics()
 }
 
 func (s *asyncSingletonAllocator) SignalDraining() error {

diff --git a/pkg/processor/eventprocessor/statistics.go b/pkg/processor/eventprocessor/statistics.go
@@ -18,34 +18,6 @@ package eventprocessor
 
 import "sync/atomic"
 
-type Statistics struct {
-	EventsHandledSuccess uint64
-	EventsHandledError   uint64
-}
-
-// AllocatorStatistics is not a safe statistics object and should be used only to copy safe object to it and return to outside
-// so it can later be taken by metric gatherers
-type AllocatorStatistics struct {
-	AllocationCount                       uint64
-	AllocationSuccessImmediateTotal       uint64
-	AllocationSuccessAfterWaitTotal       uint64
-	AllocationTimeoutTotal                uint64
-	AllocationWaitDurationMilliSecondsSum uint64
-	AllocationObjectsAvailablePercentage  uint64
-}
-
-func (s *AllocatorStatistics) DiffFrom(prev *AllocatorStatistics) AllocatorStatistics {
-
-	return AllocatorStatistics{
-		AllocationCount:                       s.AllocationCount - prev.AllocationCount,
-		AllocationSuccessImmediateTotal:       s.AllocationSuccessImmediateTotal - prev.AllocationSuccessImmediateTotal,
-		AllocationSuccessAfterWaitTotal:       s.AllocationSuccessAfterWaitTotal - prev.AllocationSuccessAfterWaitTotal,
-		AllocationTimeoutTotal:                s.AllocationTimeoutTotal - prev.AllocationTimeoutTotal,
-		AllocationWaitDurationMilliSecondsSum: s.AllocationWaitDurationMilliSecondsSum - prev.AllocationWaitDurationMilliSecondsSum,
-		AllocationObjectsAvailablePercentage:  s.AllocationObjectsAvailablePercentage - prev.AllocationObjectsAvailablePercentage,
-	}
-}
-
 // safeAllocatorStatistics is a safe statistics object, for outside usages use AllocatorStatistics
 type safeAllocatorStatistics struct {
 	AllocationCount                       atomic.Uint64

diff --git a/pkg/processor/eventprocessor/types.go b/pkg/processor/eventprocessor/types.go
@@ -23,6 +23,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/processor/cloudevent"
 	"github.com/nuclio/nuclio/pkg/processor/controlcommunication"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
@@ -49,7 +50,7 @@ type Allocator interface {
 	GetNumObjectsAvailable() int
 
 	// GetStatistics returns allocator statistics
-	GetStatistics() *AllocatorStatistics
+	GetStatistics() *statistics.AllocatorStatistics
 
 	// SignalDraining signals all event processors to drain events
 	SignalDraining() error
@@ -99,7 +100,10 @@ type EventProcessor interface {
 	Stop() error
 
 	// GetStatistics returns event processing statistics, s
6DAF
uch as counts and latencies
-	GetStatistics() *Statistics
+	GetStatistics() *statistics.EventProcessingStatistics
+
+	// GetAllocationStatistics returns allocation statistics, such as allocation time and number of allocations
+	GetAllocationStatistics() *statistics.AllocatorStatistics
 
 	// GetStructuredCloudEvent retrieves the last processed structured CloudEvent
 	GetStructuredCloudEvent() *cloudevent.Structured

diff --git a/pkg/processor/runtime/golang/runtime.go b/pkg/processor/runtime/golang/runtime.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/nuclio/nuclio/pkg/common/status"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
@@ -119,6 +120,12 @@ func (g *golang) ProcessBatch(batch []nuclio.Event, functionLogger logger.Logger
 	return nil, nuclio.ErrNotImplemented
 }
 
+// GetAllocationStatistics returns the statistics of the allocator if there is any in the runtime
+// go runtime doesn't have any allocator in it, so return nil
+func (g *golang) GetAllocationStatistics() *statistics.AllocatorStatistics {
+	return nil
+}
+
 func (g *golang) Restart() error {
 	if err := g.Stop(); err != nil {
 		return errors.Wrap(err, "Failed to stop golang runtime")

diff --git a/pkg/processor/runtime/rpc/abstract.go b/pkg/processor/runtime/rpc/abstract.go
@@ -27,6 +27,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/processor/eventprocessor"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
 	"github.com/nuclio/nuclio/pkg/processor/runtime/rpc/connection"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 	"github.com/nuclio/nuclio/pkg/processwaiter"
 
 	"github.com/nuclio/errors"
@@ -90,6 +91,10 @@ func (r *AbstractRuntime) ProcessBatch(batch []nuclio.Event, functionLogger logg
 	return r.processBatchAndWaitForResult(batch, functionLogger)
 }
 
+func (r *AbstractRuntime) GetAllocationStatistics() *statistics.AllocatorStatistics {
+	return r.connectionManager.GetAllocationStatistics()
+}
+
 // Stop stops the runtime
 func (r *AbstractRuntime) Stop() error {
 	r.Logger.WarnWith("Stopping",

diff --git a/pkg/processor/runtime/rpc/connection/abstract.go b/pkg/processor/runtime/rpc/connection/abstract.go
@@ -37,6 +37,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/processor/runtime/rpc/controlmessagebroker"
 	"github.com/nuclio/nuclio/pkg/processor/runtime/rpc/encoder"
 	"github.com/nuclio/nuclio/pkg/processor/runtime/rpc/result"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/mitchellh/mapstructure"
 	"github.com/nuclio/errors"
@@ -88,6 +89,10 @@ func (bc *AbstractConnectionManager) UpdateStatistics(durationSec float64) {
 	bc.Configuration.Statistics.DurationMilliSecondsSum += uint64(durationSec * 1000)
 }
 
+func (bc *AbstractConnectionManager) GetAllocationStatistics() *statistics.AllocatorStatistics {
+	return bc.allocator.GetStatistics()
+}
+
 func (bc *AbstractConnectionManager) GetConfig() ManagerConfigration {
 	return *bc.Configuration
 }
@@ -468,7 +473,13 @@ func (be *AbstractEventConnection) RunHandler() {
 }
 
 // GetStatistics returns a pointer to the statistics object. This must not be modified by the reader
-func (be *AbstractEventConnection) GetStatistics() *eventprocessor.Statistics {
+func (be *AbstractEventConnection) GetStatistics() *statistics.EventProcessingStatistics {
+	return nil
+}
+
+// GetAllocationStatistics returns the statistics of the allocator if there is any in the runtime
+// AbstractEventConnection runtime doesn't have any allocator in it, so return nil
+func (be *AbstractEventConnection) GetAllocationStatistics() *statistics.AllocatorStatistics {
 	return nil
 }
 

diff --git a/pkg/processor/runtime/rpc/connection/types.go b/pkg/processor/runtime/rpc/connection/types.go
@@ -26,6 +26,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
 	"github.com/nuclio/nuclio/pkg/processor/runtime/rpc/encoder"
 	"github.com/nuclio/nuclio/pkg/processor/runtime/rpc/result"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/logger"
 )
@@ -54,6 +55,9 @@ type ConnectionManager interface {
 	// duration of an event or process, specified in seconds
 	UpdateStatistics(durationSec float64)
 
+	// GetAllocationStatistics retrieves the current statistics of the ConnectionManager allocator
+	GetAllocationStatistics() *statistics.AllocatorStatistics
+
 	// SetStatus updates the operational status of the ConnectionManager
 	SetStatus(status.Status)
 

diff --git a/pkg/processor/runtime/runtime.go b/pkg/processor/runtime/runtime.go
@@ -23,6 +23,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/common/status"
 	"github.com/nuclio/nuclio/pkg/processor/controlcommunication"
 	"github.com/nuclio/nuclio/pkg/processor/databinding"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
@@ -44,6 +45,10 @@ type Runtime interface {
 	// GetStatistics returns statistics gathered by the runtime
 	GetStatistics() *Statistics
 
+	// GetAllocationStatistics returns statistics gathered by the allocator
+	// if there is any in the runtime, otherwise returns nil
+	GetAllocationStatistics() *statistics.AllocatorStatistics
+
 	// GetConfiguration returns the runtime configuration
 	GetConfiguration() *Configuration
 

diff --git a/pkg/processor/runtime/shell/runtime.go b/pkg/processor/runtime/shell/runtime.go
@@ -31,6 +31,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/common/status"
 	"github.com/nuclio/nuclio/pkg/functionconfig"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
@@ -176,6 +177,12 @@ func (s *shell) ProcessBatch(batch []nuclio.Event, functionLogger logger.Logger)
 	return nil, nuclio.ErrNotImplemented
 }
 
+// GetAllocationStatistics returns the statistics of the allocator if there is any in the runtime
+// shell runtime doesn't have any allocator in it, so return nil
+func (s *shell) GetAllocationStatistics() *statistics.AllocatorStatistics {
+	return nil
+}
+
 func (s *shell) processEvent(context context.Context,
 	command []string,
 	event nuclio.Event,

diff --git a/pkg/processor/statistics/gatherer.go → ...processor/statistics/gatherer/gatherer.go b/pkg/processor/statistics/gatherer.go → ...processor/statistics/gatherer/gatherer.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package statistics
+package gatherer
 
 // a reflection of an object in the processor (e.g. trigger, runtime, worker) that holds prometheus
 // metrics. when Gather() is called, the resource is queried for its primitive statistics. this way we decouple

diff --git a/pkg/processor/statistics/metricpusher.go → ...essor/statistics/gatherer/metricpusher.go b/pkg/processor/statistics/metricpusher.go → ...essor/statistics/gatherer/metricpusher.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package statistics
+package gatherer
 
 import (
 	"os"

diff --git a/pkg/processor/statistics/trigger.go → pkg/processor/statistics/gatherer/trigger.go b/pkg/processor/statistics/trigger.go → pkg/processor/statistics/gatherer/trigger.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package statistics
+package gatherer
 
 import (
 	"sync/atomic"

diff --git a/pkg/processor/statistics/worker.go → pkg/processor/statistics/gatherer/worker.go b/pkg/processor/statistics/worker.go → pkg/processor/statistics/gatherer/worker.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package statistics
+package gatherer
 
 import (
 	"strconv"

diff --git a/pkg/processor/statistics/types.go b/pkg/processor/statistics/types.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2025 The Nuclio Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package statistics
+
+type EventProcessingStatistics struct {
+	EventsHandledSuccess uint64
+	EventsHandledError   uint64
+}
+
+// AllocatorStatistics is not a safe statistics object and should be used only to copy safe object to it and return to outside
+// so it can later be taken by metric gatherers
+type AllocatorStatistics struct {
+	AllocationCount                       uint64
+	AllocationSuccessImmediateTotal       uint64
+	AllocationSuccessAfterWaitTotal       uint64
+	AllocationTimeoutTotal                uint64
+	AllocationWaitDurationMilliSecondsSum uint64
+	AllocationObjectsAvailablePercentage  uint64
+}
+
+func (s *AllocatorStatistics) DiffFrom(prev *AllocatorStatistics) AllocatorStatistics {
+
+	return AllocatorStatistics{
+		AllocationCount:                       s.AllocationCount - prev.AllocationCount,
+		AllocationSuccessImmediateTotal:       s.AllocationSuccessImmediateTotal - prev.AllocationSuccessImmediateTotal,
+		AllocationSuccessAfterWaitTotal:       s.AllocationSuccessAfterWaitTotal - prev.AllocationSuccessAfterWaitTotal,
+		AllocationTimeoutTotal:                s.AllocationTimeoutTotal - prev.AllocationTimeoutTotal,
+		AllocationWaitDurationMilliSecondsSum: s.AllocationWaitDurationMilliSecondsSum - prev.AllocationWaitDurationMilliSecondsSum,
+		AllocationObjectsAvailablePercentage:  s.AllocationObjectsAvailablePercentage - prev.AllocationObjectsAvailablePercentage,
+	}
+}
diff --git a/pkg/processor/trigger/types.go b/pkg/processor/trigger/types.go
@@ -23,8 +23,8 @@ import (
 	"time"
 
 	"github.com/nuclio/nuclio/pkg/functionconfig"
-	"github.com/nuclio/nuclio/pkg/processor/eventprocessor"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 	"github.com/nuclio/nuclio/pkg/processor/util/partitionworker"
 
 	"github.com/nuclio/errors"
@@ -183,7 +183,7 @@ func (c *Configuration) ResolveWorkerAllocationMode(modeFromAttributes, modeFrom
 type Statistics struct {
 	EventsHandledSuccessTotal uint64
 	EventsHandledFailureTotal uint64
-	WorkerAllocatorStatistics eventprocessor.AllocatorStatistics
+	WorkerAllocatorStatistics statistics.AllocatorStatistics
 }
 
 func (s *Statistics) DiffFrom(prev *Statistics) Statistics {

diff --git a/pkg/processor/worker/worker.go b/pkg/processor/worker/worker.go
@@ -27,6 +27,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/processor/controlcommunication"
 	"github.com/nuclio/nuclio/pkg/processor/eventprocessor"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/logger"
 	"github.com/nuclio/nuclio-sdk-go"
@@ -36,7 +37,7 @@ import (
 type Worker struct {
 
 	// accessed atomically, keep as first field for alignment
-	statistics eventprocessor.Statistics
+	statistics statistics.EventProcessingStatistics
 
 	logger               logger.Logger
 	index                int
@@ -93,10 +94,14 @@ func (w *Worker) ProcessEventBatch(batch []nuclio.Event, functionLogger logger.L
 }
 
 // GetStatistics returns a pointer to the statistics object. This must not be modified by the reader
-func (w *Worker) GetStatistics() *eventprocessor.Statistics {
+func (w *Worker) GetStatistics() *statistics.EventProcessingStatistics {
 	return &w.statistics
 }
 
+func (w *Worker) GetAllocationStatistics() *statistics.AllocatorStatistics {
+	return w.runtime.GetAllocationStatistics()
+}
+
 // GetIndex returns the index of the worker, as specified during creation
 func (w *Worker) GetIndex() int {
 	return w.index

diff --git a/pkg/processor/worker/worker_test.go b/pkg/processor/worker/worker_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/nuclio/nuclio/pkg/common/status"
 	"github.com/nuclio/nuclio/pkg/processor/controlcommunication"
 	"github.com/nuclio/nuclio/pkg/processor/runtime"
+	"github.com/nuclio/nuclio/pkg/processor/statistics"
 
 	"github.com/nuclio/logger"
 	"github.com/nuclio/nuclio-sdk-go"
@@ -53,6 +54,10 @@ func (mr *MockRuntime) GetStatistics() *runtime.Statistics {
 	return nil
 }
 
+func (mr *MockRuntime) GetAllocationStatistics() *statistics.AllocatorStatistics {
+	return nil
+}
+
 func (mr *MockRuntime) GetConfiguration() *runtime.Configuration {
 	return nil
 }