[coordinator] Introduce a unified service manager

Signed-off-by: Liran Funaro <liran.funaro@gmail.com>

Author: liran-funaro

docs/metrics_reference.md

@@ -51,22 +51,22 @@ The following Sidecar metrics are exported for consumption by Prometheus.
 
 The following Coordinator metrics are exported for consumption by Prometheus.
 
-| Name                                                       | Type    | Labels      | Description                                                                                                |
-|------------------------------------------------------------|---------|-------------|------------------------------------------------------------------------------------------------------------|
-| coordinator_grpc_received_transaction_total                | counter |             | Total number of transactions received by the coordinator service from the client.                          |
-| coordinator_grpc_committed_transaction_total               | counter | status      | Total number of transactions committed status sent by the coordinator service to the client.               |
-| coordinator_verifier_input_tx_batch_queue_size             | gauge   |             | Size of the input transaction batch queue of the signature verifier manager.                               |
-| coordinator_verifier_output_validated_tx_batch_queue_size  | gauge   |             | Size of the output validated transaction batch queue of the signature verifier manager.                    |
-| coordinator_vcservice_output_tx_status_batch_queue_size    | gauge   |             | Size of the output transaction status batch queue of the validation and committer service manager.         |
-| coordinator_vcservice_output_validated_tx_batch_queue_size | gauge   |             | Size of the output validated transaction batch queue of the validation and committer service manager.      |
-| coordinator_verifier_transaction_processed_total           | counter |             | Total number of transactions processed by the signature verifier manager.                                  |
-| coordinator_vcservice_transaction_processed_total          | counter |             | Total number of transactions processed by the validation and committer service manager.                    |
-| coordinator_verifier_connection_status                     | gauge   | grpc_target | Connection status to verifier service by grpc target (1 = connected, 0 = disconnected).                    |
-| coordinator_verifier_connection_failure_total              | counter | grpc_target | Total number of connection failures to verifier service. Short-lived failures may not always be captured.  |
-| coordinator_vcservice_retired_transaction_total            | counter |             | Total number of transactions retried by the validation and committer service manager.                      |
-| coordinator_vcservice_connection_status                    | gauge   | grpc_target | Connection status to vcservice service by grpc target (1 = connected, 0 = disconnected).                   |
-| coordinator_vcservice_connection_failure_total             | counter | grpc_target | Total number of connection failures to vcservice service. Short-lived failures may not always be captured. |
-| coordinator_verifier_retired_transaction_total             | counter |             | Total number of transactions retried by the signature verifier manager.                                    |
+| Name                                              | Type    | Labels      | Description                                                                                                |
+|---------------------------------------------------|---------|-------------|------------------------------------------------------------------------------------------------------------|
+| coordinator_grpc_received_transaction_total       | counter |             | Total number of transactions received by the coordinator service from the client.                          |
+| coordinator_grpc_committed_transaction_total      | counter | status      | Total number of transactions committed status sent by the coordinator service to the client.               |
+| coordinator_verifier_connection_status            | gauge   | grpc_target | Connection status to verifier service by grpc target (1 = connected, 0 = disconnected).                    |
+| coordinator_verifier_connection_failure_total     | counter | grpc_target | Total number of connection failures to verifier service. Short-lived failures may not always be captured.  |
+| coordinator_verifier_transaction_processed_total  | counter |             | Total number of transactions processed by the manager.                                                     |
+| coordinator_verifier_transaction_retired_total    | counter |             | Total number of transactions retried by the manager.                                                       |
+| coordinator_verifier_input_batch_queue_size       | gauge   |             | Size of the input batch queue of the manager.                                                              |
+| coordinator_verifier_output_batch_queue_size      | gauge   |             | Size of the output batch queue of the manager.                                                             |
+| coordinator_vcservice_connection_status           | gauge   | grpc_target | Connection status to vcservice service by grpc target (1 = connected, 0 = disconnected).                   |
+| coordinator_vcservice_connection_failure_total    | counter | grpc_target | Total number of connection failures to vcservice service. Short-lived failures may not always be captured. |
+| coordinator_vcservice_transaction_processed_total | counter |             | Total number of transactions processed by the manager.                                                     |
+| coordinator_vcservice_transaction_retired_total   | counter |             | Total number of transactions retried by the manager.                                                       |
+| coordinator_vcservice_input_batch_queue_size      | gauge   |             | Size of the input batch queue of the manager.                                                              |
+| coordinator_vcservice_output_batch_queue_size     | gauge   |             | Size of the output batch queue of the manager.                                                             |
 
 ## Verifier Metrics
 

scripts/metrics_doc_extract.py

@@ -64,10 +64,14 @@ def print_all_metrics_from_content(content: str):
             print_single_metric_from_block(metric_type, is_vec, params_block)
             continue
 
+        params_block = extract_block(params_block, "{}")
+        if not params_block:
+            continue
+
         # match_type == 'method'
         package_name, function_name = rest
 
-        # Determine the source file based on the package name
+        # Determine the source file based on the package name.
         source_file = repo_root_dir / "utils" / package_name / "metrics.go"
         if not source_file.exists():
             print(f"Warning: {source_file} not found", file=sys.stderr)
@@ -78,11 +82,13 @@ def print_all_metrics_from_content(content: str):
         if not func_body:
             continue
 
-        # Substitute params.Namespace and params.Subsystem in the function body
+        # Substitute "params.Namespace" and "params.Subsystem" in the function body.
         namespace = extract_field(params_block, "Namespace")
         func_body = re.sub(r'\bparams\.Namespace\b', f'"{namespace}"', func_body)
         subsystem = extract_field(params_block, "Subsystem")
         func_body = re.sub(r'\bparams\.Subsystem\b', f'"{subsystem}"', func_body)
+        # Substitute "params" with the actual params definition, so we can extract nested metrics.
+        func_body = re.sub(r'\bparams\b', f'monitoring.MetricsParameters{{{params_block}}}', func_body)
 
         # Extract metrics from the substituted function body
         print_all_metrics_from_content(func_body)

service/coordinator/coordinator.go

@@ -10,7 +10,6 @@ import (
 	"context"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/hyperledger/fabric-lib-go/common/flogging"
@@ -29,6 +28,7 @@ import (
 	"github.com/hyperledger/fabric-x-committer/utils/monitoring"
 	"github.com/hyperledger/fabric-x-committer/utils/monitoring/promutil"
 	"github.com/hyperledger/fabric-x-committer/utils/serve"
+	"github.com/hyperledger/fabric-x-committer/utils/servicemanager"
 )
 
 var logger = flogging.MustGetLogger("coordinator")
@@ -39,8 +39,9 @@ type (
 	Service struct {
 		servicepb.UnimplementedCoordinatorServer
 		dependencyMgr         *dependencygraph.Manager
-		signatureVerifierMgr  *signatureVerifierManager
-		validatorCommitterMgr *validatorCommitterManager
+		signatureVerifierMgr  *servicemanager.Manager
+		validatorCommitterMgr *servicemanager.Manager
+		validatorCommitterAPI *validatorCommitterAPI
 		policyMgr             *policyManager
 		queues                *channels
 		config                *Config
@@ -142,8 +143,8 @@ func NewCoordinatorService(c *Config) *Service {
 
 	policyMgr := newPolicyManager()
 
-	svMgr := newSignatureVerifierManager(
-		&signVerifierManagerConfig{
+	svMgr := newVerifierManager(
+		&verifierManagerParams{
 			clientConfig:             &c.Verifier,
 			incomingTxsForValidation: queues.depGraphToSigVerifierFreeTxs,
 			outgoingValidatedTxs:     queues.sigVerifierToVCServiceValidatedTxs,
@@ -184,11 +185,6 @@ func (c *Service) Run(ctx context.Context) error {
 	defer cancel()
 	g, eCtx := errgroup.WithContext(canCtx)
 
-	g.Go(func() error {
-		c.monitorQueues(eCtx)
-		return nil
-	})
-
 	g.Go(func() error {
 		logger.Info("Starting dependency graph manager")
 		c.dependencyMgr.Run(eCtx)
@@ -197,7 +193,7 @@ func (c *Service) Run(ctx context.Context) error {
 
 	g.Go(func() error {
 		logger.Info("Starting signature verifier manager")
-		if err := c.signatureVerifierMgr.run(eCtx); err != nil {
+		if err := c.signatureVerifierMgr.Run(eCtx); err != nil {
 			logger.Errorf("coordinator service stops due to an error returned by signature verifier manager: %v", err)
 			return err
 		}
@@ -206,20 +202,23 @@ func (c *Service) Run(ctx context.Context) error {
 
 	g.Go(func() error {
 		logger.Info("Starting validator committer manager")
-		if err := c.validatorCommitterMgr.run(eCtx); err != nil {
+		if err := c.validatorCommitterMgr.Run(eCtx); err != nil {
 			logger.Errorf("coordinator service stopped due "+
 				"to an error returned by validator committer manager: %v", err)
 			return err
 		}
 		return nil
 	})
 
-	if !c.validatorCommitterMgr.ready.WaitForReady(eCtx) {
-		return g.Wait()
+	var err error
+	c.validatorCommitterAPI, err = newValidatorCommitterAPI(ctx, &c.config.ValidatorCommitter, c.policyMgr)
+	if err != nil {
+		return err
 	}
+	defer c.validatorCommitterAPI.close()
 
 	// We attempt to recover the policy manager and the last committed block number from the state DB.
-	if err := c.validatorCommitterMgr.recoverPolicyManagerFromStateDB(ctx); err != nil {
+	if err = c.validatorCommitterAPI.recoverPolicyManagerFromStateDB(ctx); err != nil {
 		return err
 	}
 
@@ -250,7 +249,7 @@ func (c *Service) SetLastCommittedBlockNumber(
 	lastBlock *servicepb.BlockRef,
 ) (*emptypb.Empty, error) {
 	// Error is already wrapped with proper gRPC status code by validatorCommitterMgr.
-	return &emptypb.Empty{}, c.validatorCommitterMgr.setLastCommittedBlockNumber(ctx, lastBlock)
+	return &emptypb.Empty{}, c.validatorCommitterAPI.setLastCommittedBlockNumber(ctx, lastBlock)
 }
 
 // GetNextBlockNumberToCommit returns the next expected block number to be received by the coordinator.
@@ -259,7 +258,7 @@ func (c *Service) GetNextBlockNumberToCommit(
 	_ *emptypb.Empty,
 ) (*servicepb.BlockRef, error) {
 	// Error is already wrapped with proper gRPC status code by validatorCommitterMgr.
-	return c.validatorCommitterMgr.getNextBlockNumberToCommit(ctx)
+	return c.validatorCommitterAPI.getNextBlockNumberToCommit(ctx)
 }
 
 // GetTransactionsStatus returns the status of given transactions identifiers.
@@ -268,7 +267,7 @@ func (c *Service) GetTransactionsStatus(
 	q *committerpb.TxIDsBatch,
 ) (*committerpb.TxStatusBatch, error) {
 	// Error is already wrapped with proper gRPC status code by validatorCommitterMgr.
-	return c.validatorCommitterMgr.getTransactionsStatus(ctx, q)
+	return c.validatorCommitterAPI.getTransactionsStatus(ctx, q)
 }
 
 // NoPendingTransactionProcessing returns true when all previously submitted
@@ -417,22 +416,3 @@ func (c *Service) sendTxStatus(
 		}
 	}
 }
-
-func (c *Service) monitorQueues(ctx context.Context) {
-	// TODO: make sampling time configurable
-	ticker := time.NewTicker(100 * time.Millisecond)
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-		}
-
-		m := c.metrics
-		q := c.queues
-		promutil.SetGauge(m.sigverifierInputTxBatchQueueSize, len(q.depGraphToSigVerifierFreeTxs))
-		promutil.SetGauge(m.sigverifierOutputValidatedTxBatchQueueSize, len(q.sigVerifierToVCServiceValidatedTxs))
-		promutil.SetGauge(m.vcserviceOutputValidatedTxBatchQueueSize, len(q.vcServiceToDepGraphValidatedTxs))
-		promutil.SetGauge(m.vcserviceOutputTxStatusBatchQueueSize, q.vcServiceToCoordinatorTxStatus.len())
-	}
-}

service/coordinator/coordinator_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/hyperledger/fabric-x-committer/api/servicepb"
 	"github.com/hyperledger/fabric-x-committer/mock"
 	"github.com/hyperledger/fabric-x-committer/service/coordinator/dependencygraph"
+	"github.com/hyperledger/fabric-x-committer/service/vc"
 	"github.com/hyperledger/fabric-x-committer/service/verifier/policy"
 	"github.com/hyperledger/fabric-x-committer/utils"
 	"github.com/hyperledger/fabric-x-committer/utils/channel"
@@ -523,41 +524,6 @@ func TestCoordinatorServiceDependentOrderedTxs(t *testing.T) {
 	require.Less(t, blindV2, updateV3, "v2 writer must follow blind-write; order: %v", order)
 }
 
-func TestQueueSize(t *testing.T) {
-	t.Parallel()
-	env := newCoordinatorTestEnv(t, &testConfig{numSigService: 2, numVcService: 2})
-	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Minute)
-	t.Cleanup(cancel)
-	go env.coordinator.monitorQueues(ctx)
-
-	q := env.coordinator.queues
-	m := env.coordinator.metrics
-	q.depGraphToSigVerifierFreeTxs <- dependencygraph.TxNodeBatch{}
-	q.sigVerifierToVCServiceValidatedTxs <- dependencygraph.TxNodeBatch{}
-	q.vcServiceToDepGraphValidatedTxs <- dependencygraph.TxNodeBatch{}
-	require.True(t, q.vcServiceToCoordinatorTxStatus.write(t.Context(), &committerpb.TxStatusBatch{}))
-
-	require.Eventually(t, func() bool {
-		return test.GetIntMetricValue(t, m.sigverifierInputTxBatchQueueSize) == 1 &&
-			test.GetIntMetricValue(t, m.sigverifierOutputValidatedTxBatchQueueSize) == 1 &&
-			test.GetIntMetricValue(t, m.vcserviceOutputValidatedTxBatchQueueSize) == 1 &&
-			test.GetIntMetricValue(t, m.vcserviceOutputTxStatusBatchQueueSize) == 1
-	}, 3*time.Second, 500*time.Millisecond)
-
-	<-q.depGraphToSigVerifierFreeTxs
-	<-q.sigVerifierToVCServiceValidatedTxs
-	<-q.vcServiceToDepGraphValidatedTxs
-	_, ok := q.vcServiceToCoordinatorTxStatus.read(t.Context())
-	require.True(t, ok)
-
-	require.Eventually(t, func() bool {
-		return test.GetIntMetricValue(t, m.sigverifierInputTxBatchQueueSize) == 0 &&
-			test.GetIntMetricValue(t, m.sigverifierOutputValidatedTxBatchQueueSize) == 0 &&
-			test.GetIntMetricValue(t, m.vcserviceOutputValidatedTxBatchQueueSize) == 0 &&
-			test.GetIntMetricValue(t, m.vcserviceOutputTxStatusBatchQueueSize) == 0
-	}, 3*time.Second, 500*time.Millisecond)
-}
-
 func TestCoordinatorRecovery(t *testing.T) {
 	t.Parallel()
 	env := newCoordinatorTestEnv(t, &testConfig{numSigService: 1, numVcService: 1})

service/coordinator/dependencygraph/transaction_node.go

@@ -46,7 +46,7 @@ type (
 		// Note that a transaction T2 can depend on another transaction T1 only if T1
 		// has arrived before T2 or T1 precedes T2 in transaction order in the block.
 		// Hence, we will not have cyclic dependencies.
-		dependsOnTxs TxNodeBatch
+		dependsOnTxs []*TransactionNode
 		// dependentTxs is a set of transactions that depend on this transaction.
 		// After validating this transaction, dependentTxs is used to remove dependencies
 		// from each dependent transaction.
@@ -59,7 +59,7 @@ type (
 	}
 
 	// TxNodeBatch is a batch of transaction nodes.
-	TxNodeBatch []*TransactionNode
+	TxNodeBatch = []*TransactionNode
 
 	// readWriteKeys holds the read and write keys of a transaction.
 	readWriteKeys struct {