tools: update block-generator to use conduit binary

tools/block-generator/data/conduit.yml

@@ -0,0 +1,61 @@
+# Log verbosity: PANIC, FATAL, ERROR, WARN, INFO, DEBUG, TRACE
+log-level: {{.LogLevel}}
+
+# If no log file is provided logs are written to stdout.
+log-file: {{.LogFile}}
+
+# Number of retries to perform after a pipeline plugin error.
+retry-count: 10
+
+# Time duration to wait between retry attempts.
+retry-delay: "1s"
+
+# Optional filepath to use for pidfile.
+#pid-filepath: /path/to/pidfile
+
+# Whether or not to print the conduit banner on startup.
+hide-banner: false
+
+# When enabled prometheus metrics are available on '/metrics'
+metrics:
+  mode: ON
+  addr: "{{.MetricsPort}}"
+  prefix: "conduit"
+
+# The importer is typically an algod follower node.
+importer:
+  name: algod
+  config:
+    # The mode of operation, either "archival" or "follower".
+    # * archival mode allows you to start processing on any round but does not
+    # contain the ledger state delta objects required for the postgres writer.
+    # * follower mode allows you to use a lightweight non-archival node as the
+    # data source. In addition, it will provide ledger state delta objects to
+    # the processors and exporter.
+    mode: "follower"
+
+    # Algod API address.
+    netaddr: "{{.AlgodNet}}"
+
+    # Algod API token.
+    token: ""
+
+
+# Zero or more processors may be defined to manipulate what data
+# reaches the exporter.
+processors:
+
+# An exporter is defined to do something with the data.
+exporter:
+  name: postgresql
+  config:
+    # Pgsql connection string
+    # See https://github.com/jackc/pgconn for more details
+    connection-string: "{{ .PostgresConnectionString }}"
+
+    # Maximum connection number for connection pool
+    # This means the total number of active queries that can be running
+    # concurrently can never be more than this
+    max-conn: 20
+
+

tools/block-generator/generator/generate.go

@@ -72,7 +72,7 @@
 	GenesisAccountInitialBalance uint64 `yaml:"genesis_account_balance"`
 
 	// Block generation
-	TxnPerBlock uint64 `mapstructure:"tx_per_block"`
+	TxnPerBlock uint64 `yaml:"tx_per_block"`
 
 	// TX Distribution
 	PaymentTransactionFraction float32 `yaml:"tx_pay_fraction"`
@@ -120,7 +120,7 @@
 		genesisHash:               [32]byte{},
 		genesisID:                 "blockgen-test",
 		prevBlockHash:             "",
-		round:                     1,
+		round:                     0,
 		txnCounter:                0,
 		timestamp:                 0,
 		rewardsLevel:              0,
@@ -312,7 +312,9 @@
 		FeeSink:     g.feeSink.String(),
 		Timestamp:   g.timestamp,
 	}
-	return json.NewEncoder(output).Encode(gen)
+
+	_, err := output.Write(protocol.EncodeJSON(gen))
+	return err
 }
 
 func getTransactionOptions() []interface{} {
@@ -357,12 +359,25 @@
 
 // WriteBlock generates a block full of new transactions and writes it to the writer.
 func (g *generator) WriteBlock(output io.Writer, round uint64) error {
+
 	if round != g.round {
 		fmt.Printf("Generator only supports sequential block access. Expected %d but received request for %d.\n", g.round, round)
 	}
 
 	numTxnForBlock := g.txnForRound(round)
 
+	// return genesis block
+	if round == 0 {
+		// write the msgpack bytes for a block
+		block, err := rpcs.RawBlockBytes(g.ledger, basics.Round(round))
+		if err != nil {
+			return err
+		}
+		output.Write(block)
+		g.finishRound(numTxnForBlock)
+		return nil
+	}
+
 	header := bookkeeping.BlockHeader{
 		Round:          basics.Round(g.round),
 		Branch:         bookkeeping.BlockHash{},
@@ -414,15 +429,16 @@
 		Certificate: agreement.Certificate{},
 	}
 
-	err := json.NewEncoder(output).Encode(cert)
+	err := g.ledger.AddBlock(cert.Block, cert.Certificate)
 	if err != nil {
 		return err
 	}
-	err = g.ledger.AddBlock(cert.Block, agreement.Certificate{})
+	// write the msgpack bytes for a block
+	block, err := rpcs.RawBlockBytes(g.ledger, basics.Round(round))
 	if err != nil {
 		return err
 	}
-	g.ledger.WaitForCommit(basics.Round(g.round))
+	output.Write(block)
 	g.finishRound(numTxnForBlock)
 	return nil
 }

tools/block-generator/generator/server_test.go

@@ -53,61 +53,61 @@ func TestParseURL(t *testing.T) {
 	var testcases = []struct {
 		name          string
 		url           string
-		expectedRound string
+		expectedParam string
 		err           string
 	}{
 		{
 			name:          "no block",
 			url:           "/v2/blocks/",
-			expectedRound: "",
+			expectedParam: "",
 			err:           "invalid request path, /v2/blocks/",
 		},
 		{
 			name:          "normal one digit",
 			url:           fmt.Sprintf("%s1", blockQueryPrefix),
-			expectedRound: "1",
+			expectedParam: "1",
 			err:           "",
 		},
 		{
 			name:          "normal long number",
 			url:           fmt.Sprintf("%s12345678", blockQueryPrefix),
-			expectedRound: "12345678",
+			expectedParam: "12345678",
 			err:           "",
 		},
 		{
 			name:          "with query parameters",
 			url:           fmt.Sprintf("%s1234?pretty", blockQueryPrefix),
-			expectedRound: "1234",
+			expectedParam: "1234",
 			err:           "",
 		},
 		{
 			name:          "with query parameters",
 			url:           fmt.Sprintf("%s1234?pretty", blockQueryPrefix),
-			expectedRound: "1234",
+			expectedParam: "1234",
 			err:           "",
 		},
 		{
 			name:          "no deltas",
 			url:           "/v2/deltas/",
-			expectedRound: "",
+			expectedParam: "",
 			err:           "invalid request path, /v2/deltas/",
 		},
 		{
 			name:          "deltas",
 			url:           fmt.Sprintf("%s123?Format=msgp", deltaQueryPrefix),
-			expectedRound: "123",
+			expectedParam: "123",
 			err:           "",
 		},
 		{
 			name:          "no account",
 			url:           "/v2/accounts/",
-			expectedRound: "",
+			expectedParam: "",
 			err:           "invalid request path, /v2/accounts/",
 		},
 		{
 			name:          "accounts",
 			url:           fmt.Sprintf("%sAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGFFWAF4", accountQueryPrefix),
-			expectedRound: "AIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGFFWAF4",
+			expectedParam: "AIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGFFWAF4",
 			err:           "",
 		},
 	}
@@ -117,9 +117,9 @@ func TestParseURL(t *testing.T) {
 			round, err := parseURL(testcase.url)
 			if len(testcase.err) == 0 {
 				msg := fmt.Sprintf("Unexpected error parsing '%s', expected round '%s' received error: %v",
-					testcase.url, testcase.expectedRound, err)
+					testcase.url, testcase.expectedParam, err)
 				require.NoError(t, err, msg)
-				assert.Equal(t, testcase.expectedRound, round)
+				assert.Equal(t, testcase.expectedParam, round)
 			} else {
 				require.Error(t, err, fmt.Sprintf("Expected an error containing: %s", testcase.err))
 				require.True(t, strings.Contains(err.Error(), testcase.err))

tools/block-generator/run_tests.sh

@@ -17,7 +17,7 @@ help() {
   echo " -r|--report-dir directory where the report should be written."
   echo " -d|--duration   test duration."
   echo " -l|--level      log level to pass to Indexer."
-  echo " -g|--generator  use a different indexer binary to run the generator."
+  echo " -g|--generator  use a different block-generator binary to run the generator."
   exit
 }
 

tools/block-generator/runner/metrics_collector.go

@@ -24,6 +24,31 @@ import (
 	"time"
 )
 
+// Prometheus metrics collected in Conduit.
+const (
+	BlockImportTimeName      = "import_time_sec"
+	ImportedTxnsPerBlockName = "imported_tx_per_block"
+	ImportedRoundGaugeName   = "imported_round"
+	GetAlgodRawBlockTimeName = "get_algod_raw_block_time_sec"
+	ImportedTxnsName         = "imported_txns"
+	ImporterTimeName         = "importer_time_sec"
+	ProcessorTimeName        = "processor_time_sec"
+	ExporterTimeName         = "exporter_time_sec"
+	PipelineRetryCountName   = "pipeline_retry_count"
+)
+
+// AllMetricNames is a reference for all the custom metric names.
+var AllMetricNames = []string{
+	BlockImportTimeName,
+	ImportedTxnsPerBlockName,
+	ImportedRoundGaugeName,
+	GetAlgodRawBlockTimeName,
+	ImporterTimeName,
+	ProcessorTimeName,
+	ExporterTimeName,
+	PipelineRetryCountName,
+}
+
 // MetricsCollector queries a /metrics endpoint for prometheus style metrics and saves metrics matching a pattern.
 type MetricsCollector struct {
 	// MetricsURL where metrics can be queried.