file-plugins: enabling messagepack format

conduit/plugins/exporters/filewriter/README.md

@@ -2,7 +2,10 @@
 
 Write block data to files. This plugin works with the file rerader plugin to create a simple file-based pipeine.
 
+The genesis is always exported to a plain JSON file named `genesis.json` regardless of the `FilenamePattern`.
+
 ## Configuration
+
 ```yml @sample.yaml
 name: file_writer
 config:
@@ -13,9 +16,11 @@ config:
 
     # FilenamePattern is the format used to write block files. It uses go
     # string formatting and should accept one number for the round.
-    # If the file has a '.gz' extension, blocks will be gzipped.
-    # Default: "%[1]d_block.json"
-    filename-pattern: "%[1]d_block.json"
+    # To specify JSON encoding, add a '.json' extension to the filename.
+    # To specify MessagePack encoding, add a '.msgp' extension to the filename.
+    # If the file has a '.gz' extension, blocks will be gzipped regardless of encoding.
+    # Default: "%[1]d_block.msgp.gz"
+    filename-pattern: "%[1]d_block.msgp.gz"
 
     # DropCertificate is used to remove the vote certificate from the block data before writing files.
     drop-certificate: true

conduit/plugins/exporters/filewriter/file_exporter.go

@@ -18,13 +18,19 @@ import (
 const (
 	// PluginName to use when configuring.
 	PluginName = "file_writer"
+
 	// FilePattern is used to name the output files.
-	FilePattern = "%[1]d_block.json"
+	FilePattern = "%[1]d_block.msgp.gz"
+
+	// GenesisFilename is the name of the genesis file.
+	GenesisFilename = "genesis.json"
 )
 
 type fileExporter struct {
 	round  uint64
 	cfg    Config
+	gzip   bool
+	format EncodingFormat
 	logger *logrus.Logger
 }
 
@@ -51,20 +57,34 @@ func (exp *fileExporter) Init(_ context.Context, initProvider data.InitProvider,
 	if exp.cfg.FilenamePattern == "" {
 		exp.cfg.FilenamePattern = FilePattern
 	}
+	exp.format, exp.gzip, err = ParseFilenamePattern(exp.cfg.FilenamePattern)
+	if err != nil {
+		return fmt.Errorf("Init() error: %w", err)
+	}
+
 	// default to the data directory if no override provided.
 	if exp.cfg.BlocksDir == "" {
 		exp.cfg.BlocksDir = cfg.DataDir
 	}
 	// create block directory
 	err = os.Mkdir(exp.cfg.BlocksDir, 0755)
-	if err != nil && errors.Is(err, os.ErrExist) {
-		// Ignore mkdir if the dir exists
-		err = nil
-	} else if err != nil {
+	if err != nil && !errors.Is(err, os.ErrExist) {
+		// Ignore mkdir err if the dir exists (case errors.Is(err, os.ErrExist))
 		return fmt.Errorf("Init() error: %w", err)
 	}
+
 	exp.round = uint64(initProvider.NextDBRound())
-	return err
+
+	genesis := initProvider.GetGenesis()
+	genesisPath := path.Join(exp.cfg.BlocksDir, GenesisFilename)
+
+	// the genesis is always exported as plain JSON:
+	err = EncodeToFile(genesisPath, genesis, JSONFormat, false)
+	if err != nil {
+		return fmt.Errorf("Init() error sending to genesisPath=%s: %w", genesisPath, err)
+	}
+
+	return nil
 }
 
 func (exp *fileExporter) Close() error {
@@ -87,10 +107,12 @@ func (exp *fileExporter) Receive(exportData data.BlockData) error {
 		}
 
 		blockFile := path.Join(exp.cfg.BlocksDir, fmt.Sprintf(exp.cfg.FilenamePattern, exportData.Round()))
-		err := EncodeJSONToFile(blockFile, exportData, true)
+
+		err := EncodeToFile(blockFile, &exportData, exp.format, exp.gzip)
 		if err != nil {
 			return fmt.Errorf("Receive(): failed to write file %s: %w", blockFile, err)
 		}
+
 		exp.logger.Infof("Wrote block %d to %s", exportData.Round(), blockFile)
 	}
 

conduit/plugins/exporters/filewriter/file_exporter_test.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 
@@ -21,29 +20,47 @@ import (
 	sdk "github.com/algorand/go-algorand-sdk/v2/types"
 )
 
+const (
+	defaultEncodingFormat = MessagepackFormat
+	defaultIsGzip         = true
+)
+
 var logger *logrus.Logger
 var fileCons = exporters.ExporterConstructorFunc(func() exporters.Exporter {
 	return &fileExporter{}
 })
-var configTemplate = "block-dir: %s/blocks\n"
+var configTemplatePrefix = "block-dir: %s/blocks\n"
 var round = sdk.Round(2)
 
 func init() {
 	logger, _ = test.NewNullLogger()
 }
 
-func getConfig(t *testing.T) (config, tempdir string) {
+func getConfigWithoutPattern(t *testing.T) (config, tempdir string) {
 	tempdir = t.TempDir()
-	config = fmt.Sprintf(configTemplate, tempdir)
+	config = fmt.Sprintf(configTemplatePrefix, tempdir)
+	return
+}
+
+func getConfigWithPattern(t *testing.T, pattern string) (config, tempdir string) {
+	config, tempdir = getConfigWithoutPattern(t)
+	config = fmt.Sprintf("%sfilename-pattern: '%s'\n", config, pattern)
 	return
 }
 
+func TestDefaults(t *testing.T) {
+	format, gzip, err := ParseFilenamePattern(FilePattern)
+	require.NoError(t, err)
+	require.Equal(t, format, defaultEncodingFormat)
+	require.Equal(t, gzip, defaultIsGzip)
+}
+
 func TestExporterMetadata(t *testing.T) {
 	fileExp := fileCons.New()
 	meta := fileExp.Metadata()
-	assert.Equal(t, metadata.Name, meta.Name)
-	assert.Equal(t, metadata.Description, meta.Description)
-	assert.Equal(t, metadata.Deprecated, meta.Deprecated)
+	require.Equal(t, metadata.Name, meta.Name)
+	require.Equal(t, metadata.Description, meta.Description)
+	require.Equal(t, metadata.Deprecated, meta.Deprecated)
 }
 
 func TestExporterInitDefaults(t *testing.T) {
@@ -87,18 +104,18 @@ func TestExporterInitDefaults(t *testing.T) {
 }
 
 func TestExporterInit(t *testing.T) {
-	config, _ := getConfig(t)
+	config, _ := getConfigWithPattern(t, "%[1]d_block.json")
 	fileExp := fileCons.New()
 	defer fileExp.Close()
 
 	// creates a new output file
 	err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), plugins.MakePluginConfig(config), logger)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	fileExp.Close()
 
 	// can open existing file
 	err = fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), plugins.MakePluginConfig(config), logger)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	fileExp.Close()
 }
 
@@ -155,55 +172,67 @@ func sendData(t *testing.T, fileExp exporters.Exporter, config string, numRounds
 }
 
 func TestExporterReceive(t *testing.T) {
-	config, tempdir := getConfig(t)
-	fileExp := fileCons.New()
-	numRounds := 5
-	sendData(t, fileExp, config, numRounds)
-
-	// block data is valid
-	for i := 0; i < 5; i++ {
-		filename := fmt.Sprintf(FilePattern, i)
-		path := fmt.Sprintf("%s/blocks/%s", tempdir, filename)
-		require.FileExists(t, path)
-
-		blockBytes, err := os.ReadFile(path)
-		require.NoError(t, err)
-		assert.NotContains(t, string(blockBytes), " 0: ")
+	patterns := []string{
+		"%[1]d_block.json",
+		"%[1]d_block.json.gz",
+		"%[1]d_block.msgp",
+		"%[1]d_block.msgp.gz",
+	}
+	for _, pattern := range patterns {
+		pattern := pattern
+		t.Run(pattern, func(t *testing.T) {
+			t.Parallel()
 
-		var blockData data.BlockData
-		err = DecodeJSONFromFile(path, &blockData, true)
-		require.Equal(t, sdk.Round(i), blockData.BlockHeader.Round)
-		require.NoError(t, err)
-		require.NotNil(t, blockData.Certificate)
+			format, isGzip, err := ParseFilenamePattern(pattern)
+			require.NoError(t, err)
+			config, tempdir := getConfigWithPattern(t, pattern)
+			fileExp := fileCons.New()
+			numRounds := 5
+			sendData(t, fileExp, config, numRounds)
+
+			// block data is valid
+			for i := 0; i < 5; i++ {
+				filename := fmt.Sprintf(pattern, i)
+				path := fmt.Sprintf("%s/blocks/%s", tempdir, filename)
+				require.FileExists(t, path)
+
+				blockBytes, err := os.ReadFile(path)
+				require.NoError(t, err)
+				require.NotContains(t, string(blockBytes), " 0: ")
+
+				var blockData data.BlockData
+				err = DecodeFromFile(path, &blockData, format, isGzip)
+				require.NoError(t, err)
+				require.Equal(t, sdk.Round(i), blockData.BlockHeader.Round)
+				require.NotNil(t, blockData.Certificate)
+			}
+		})
 	}
 }
 
 func TestExporterClose(t *testing.T) {
-	config, _ := getConfig(t)
+	config, _ := getConfigWithoutPattern(t)
 	fileExp := fileCons.New()
 	rnd := sdk.Round(0)
 	fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&rnd, nil, nil), plugins.MakePluginConfig(config), logger)
 	require.NoError(t, fileExp.Close())
 }
 
-func TestPatternOverride(t *testing.T) {
-	config, tempdir := getConfig(t)
+func TestPatternDefault(t *testing.T) {
+	config, tempdir := getConfigWithoutPattern(t)
 	fileExp := fileCons.New()
 
-	patternOverride := "PREFIX_%[1]d_block.json"
-	config = fmt.Sprintf("%sfilename-pattern: '%s'\n", config, patternOverride)
-
 	numRounds := 5
 	sendData(t, fileExp, config, numRounds)
 
 	// block data is valid
 	for i := 0; i < 5; i++ {
-		filename := fmt.Sprintf(patternOverride, i)
+		filename := fmt.Sprintf(FilePattern, i)
 		path := fmt.Sprintf("%s/blocks/%s", tempdir, filename)
-		assert.FileExists(t, path)
+		require.FileExists(t, path)
 
 		var blockData data.BlockData
-		err := DecodeJSONFromFile(path, &blockData, true)
+		err := DecodeFromFile(path, &blockData, defaultEncodingFormat, defaultIsGzip)
 		require.Equal(t, sdk.Round(i), blockData.BlockHeader.Round)
 		require.NoError(t, err)
 		require.NotNil(t, blockData.Certificate)
@@ -227,10 +256,10 @@ func TestDropCertificate(t *testing.T) {
 	for i := 0; i < numRounds; i++ {
 		filename := fmt.Sprintf(FilePattern, i)
 		path := fmt.Sprintf("%s/%s", tempdir, filename)
-		assert.FileExists(t, path)
+		require.FileExists(t, path)
 		var blockData data.BlockData
-		err := DecodeJSONFromFile(path, &blockData, true)
-		assert.NoError(t, err)
-		assert.Nil(t, blockData.Certificate)
+		err := DecodeFromFile(path, &blockData, defaultEncodingFormat, defaultIsGzip)
+		require.NoError(t, err)
+		require.Nil(t, blockData.Certificate)
 	}
 }

conduit/plugins/exporters/filewriter/sample.yaml

@@ -7,9 +7,11 @@ config:
 
     # FilenamePattern is the format used to write block files. It uses go
     # string formatting and should accept one number for the round.
-    # If the file has a '.gz' extension, blocks will be gzipped.
-    # Default: "%[1]d_block.json"
-    filename-pattern: "%[1]d_block.json"
+    # To specify JSON encoding, add a '.json' extension to the filename.
+    # To specify MessagePack encoding, add a '.msgp' extension to the filename.
+    # If the file has a '.gz' extension, blocks will be gzipped regardless of encoding.
+    # Default: "%[1]d_block.msgp.gz"
+    filename-pattern: "%[1]d_block.msgp.gz"
 
     # DropCertificate is used to remove the vote certificate from the block data before writing files.
     drop-certificate: true