Introduce support for MSP load generation

Signed-off-by: Liran Funaro <[email protected]>

Author: liran-funaro

cmd/config/app_config_test.go

@@ -379,20 +379,20 @@ func TestReadConfigLoadGen(t *testing.T) {
 				Block: workload.BlockProfile{Size: 500},
 				Transaction: workload.TransactionProfile{
 					ReadWriteCount: workload.NewConstantDistribution(2),
-					Policy: &workload.PolicyProfile{
-						ChannelID: "mychannel",
-						NamespacePolicies: map[string]*workload.Policy{
-							workload.DefaultGeneratedNamespaceID: {
-								Scheme: signature.Ecdsa, Seed: 10,
-							},
-							committerpb.MetaNamespaceID: {
-								Scheme: signature.Ecdsa, Seed: 11,
-							},
+				},
+				Policy: workload.PolicyProfile{
+					ChannelID: "mychannel",
+					NamespacePolicies: map[string]*workload.Policy{
+						workload.DefaultGeneratedNamespaceID: {
+							Scheme: signature.Ecdsa, Seed: 10,
 						},
-						OrdererEndpoints: []*commontypes.OrdererEndpoint{
-							newOrdererEndpoint("org", "orderer"),
+						committerpb.MetaNamespaceID: {
+							Scheme: signature.Ecdsa, Seed: 11,
 						},
 					},
+					OrdererEndpoints: []*commontypes.OrdererEndpoint{
+						newOrdererEndpoint("org", "orderer"),
+					},
 				},
 				Conflicts: workload.ConflictProfile{
 					InvalidSignatures: 0.1,

cmd/config/samples/loadgen.yaml

@@ -48,17 +48,17 @@ load-profile:
   transaction:
     read-write-count:
       const: 2
-    policy:
-      channel-id: mychannel
-      namespace-policies:
-        0:
-          scheme: ECDSA
-          seed: 10
-        _meta:
-          scheme: ECDSA
-          seed: 11
-      orderer-endpoints:
-        - id=0,msp-id=org,broadcast,deliver,orderer:7050
+  policy:
+    channel-id: mychannel
+    namespace-policies:
+      0:
+        scheme: ECDSA
+        seed: 10
+      _meta:
+        scheme: ECDSA
+        seed: 11
+    orderer-endpoints:
+      - id=0,msp-id=org,broadcast,deliver,orderer:7050
   conflicts:
     invalid-signatures: 0.1
   seed: 12345

cmd/config/templates/loadgen_common.yaml

@@ -38,15 +38,15 @@ load-profile:
     read-only-count:
     read-write-count:
       const: 2
-    policy:
-      channel-id: {{ .Policy.ChannelID }}
-      namespace-policies:
-        {{ range $nsID, $element := .Policy.NamespacePolicies }}
-        {{ $nsID }}:
-          scheme: {{ $element.Scheme }}
-          seed: {{ $element.Seed }}
-        {{ end }}
-      config-block-path: {{ .ConfigBlockPath }}
+  policy:
+    channel-id: {{ .Policy.ChannelID }}
+    namespace-policies:
+      {{ range $nsID, $element := .Policy.NamespacePolicies }}
+      {{ $nsID }}:
+        scheme: {{ $element.Scheme }}
+        seed: {{ $element.Seed }}
+      {{ end }}
+    config-block-path: {{ .ConfigBlockPath }}
   query:
     query-size:
     min-invalid-keys-portion:

cmd/loadgen/main.go

@@ -100,7 +100,7 @@ func loadGenGenesisBlock() *cobra.Command {
 			}
 			cmd.SilenceUsage = true
 
-			block, err := workload.CreateConfigBlock(conf.LoadProfile.Transaction.Policy)
+			block, err := workload.CreateConfigBlock(&conf.LoadProfile.Policy)
 			if err != nil {
 				return err
 			}

docker/test/container_release_image_test.go

@@ -83,7 +83,8 @@ func TestCommitterReleaseImagesWithTLS(t *testing.T) {
 	v := config.NewViperWithLoadGenDefaults()
 	c, err := config.ReadLoadGenYamlAndSetupLogging(v, filepath.Join(localConfigPath, "loadgen.yaml"))
 	require.NoError(t, err)
-	configBlock, err := workload.CreateConfigBlock(c.LoadProfile.Transaction.Policy)
+	c.LoadProfile.Policy.CryptoMaterialPath = t.TempDir()
+	configBlock, err := workload.CreateConfigBlock(&c.LoadProfile.Policy)
 	require.NoError(t, err)
 	require.NoError(t, configtxgen.WriteOutputBlock(configBlock, configBlockPath))
 

integration/runner/runtime.go

@@ -135,8 +135,9 @@ func NewRuntime(t *testing.T, conf *Config) *CommitterRuntime {
 			LoadGenBlockLimit: conf.LoadgenBlockLimit,
 			LoadGenWorkers:    1,
 			Policy: &workload.PolicyProfile{
-				ChannelID:         TestChannelName,
-				NamespacePolicies: make(map[string]*workload.Policy),
+				ChannelID:          TestChannelName,
+				NamespacePolicies:  make(map[string]*workload.Policy),
+				CryptoMaterialPath: t.TempDir(),
 			},
 			Logging: &logging.DefaultConfig,
 		},

integration/test/config_update_test.go

@@ -86,7 +86,8 @@ func TestConfigUpdate(t *testing.T) {
 	lgMetaTx := c.TxBuilder.MakeTx(metaTx)
 
 	c.AddOrUpdateNamespaces(t, committerpb.MetaNamespaceID)
-	metaPolicy := c.TxBuilder.TxEndorser.Policy(committerpb.MetaNamespaceID).VerificationPolicy()
+	verPolicies := c.TxBuilder.TxEndorser.VerificationPolicies()
+	metaPolicy := verPolicies[committerpb.MetaNamespaceID]
 	submitConfigBlock := func(endpoints []*commontypes.OrdererEndpoint) {
 		ordererEnv.SubmitConfigBlock(t, &workload.ConfigBlock{
 			ChannelID:                    c.SystemConfig.Policy.ChannelID,

integration/test/namespace_lifecycle_test.go

@@ -29,9 +29,10 @@ func TestCreateUpdateNamespace(t *testing.T) {
 	})
 	c.Start(t, runner.FullTxPath)
 
-	policyBytesNs1, err := proto.Marshal(c.TxBuilder.TxEndorser.Policy("1").VerificationPolicy())
+	verPolicies := c.TxBuilder.TxEndorser.VerificationPolicies()
+	policyBytesNs1, err := proto.Marshal(verPolicies["1"])
 	require.NoError(t, err)
-	policyBytesNs2, err := proto.Marshal(c.TxBuilder.TxEndorser.Policy("2").VerificationPolicy())
+	policyBytesNs2, err := proto.Marshal(verPolicies["2"])
 	require.NoError(t, err)
 
 	tests := []struct {

loadgen/adapters/common.go

@@ -12,6 +12,8 @@ import (
 
 	"github.com/cockroachdb/errors"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+
 	"github.com/hyperledger/fabric-x-committer/api/servicepb"
 	"github.com/hyperledger/fabric-x-committer/loadgen/metrics"
 	"github.com/hyperledger/fabric-x-committer/loadgen/workload"
@@ -23,10 +25,11 @@ import (
 type (
 	// ClientResources holds client's pre-generated resources to be used by the adapters.
 	ClientResources struct {
-		Metrics *metrics.PerfMetrics
-		Profile *workload.Profile
-		Stream  *workload.StreamOptions
-		Limit   *GenerateLimit
+		Metrics     *metrics.PerfMetrics
+		Profile     *workload.Profile
+		Stream      *workload.StreamOptions
+		ConfigBlock *common.Block
+		Limit       *GenerateLimit
 	}
 
 	// Phases specify the generation phases to enable.

loadgen/adapters/sidecar.go

@@ -28,11 +28,20 @@ type (
 )
 
 // NewSidecarAdapter instantiate SidecarAdapter.
-func NewSidecarAdapter(config *SidecarClientConfig, res *ClientResources) *SidecarAdapter {
+func NewSidecarAdapter(config *SidecarClientConfig, res *ClientResources) (*SidecarAdapter, error) {
+	// The sidecar adapter overwrite the orderer endpoints with its own.
+	// We first pre-allocate the ports.
+	for _, s := range config.OrdererServers {
+		_, err := s.PreAllocateListener()
+		if err != nil {
+			return nil, err
+		}
+	}
+	res.Profile.Policy.OrdererEndpoints = ordererconn.NewEndpoints(0, "msp", config.OrdererServers...)
 	return &SidecarAdapter{
 		commonAdapter: commonAdapter{res: res},
 		config:        config,
-	}
+	}, nil
 }
 
 // RunWorkload applies load on the sidecar.
@@ -56,13 +65,7 @@ func (c *SidecarAdapter) RunWorkload(ctx context.Context, txStream *workload.Str
 	})
 
 	// The sidecar adapter submits a config block manually.
-	policy := *c.res.Profile.Transaction.Policy
-	policy.OrdererEndpoints = ordererconn.NewEndpoints(0, "msp", c.config.OrdererServers...)
-	configBlock, err := workload.CreateConfigBlock(&policy)
-	if err != nil {
-		return err
-	}
-	if err = orderer.SubmitBlock(ctx, configBlock); err != nil {
+	if err = orderer.SubmitBlock(ctx, c.res.ConfigBlock); err != nil {
 		return err
 	}
 	c.nextBlockNum.Add(1)