Conduit: filereader / filewriter plugins and compatibility changes.

data/block_export_data.go

@@ -26,16 +26,16 @@ type InitProvider interface {
 type BlockData struct {
 
 	// BlockHeader is the immutable header from the block
-	BlockHeader bookkeeping.BlockHeader
+	BlockHeader bookkeeping.BlockHeader `json:"block,omitempty"`
 
 	// Payset is the set of data the block is carrying--can be modified as it is processed
-	Payset []transactions.SignedTxnInBlock
+	Payset []transactions.SignedTxnInBlock `json:"payset,omitempty"`
 
 	// Delta contains a list of account changes resulting from the block. Processor plugins may have modify this data.
-	Delta *ledgercore.StateDelta
+	Delta *ledgercore.StateDelta `json:"delta,omitempty"`
 
 	// Certificate contains voting data that certifies the block. The certificate is non deterministic, a node stops collecting votes once the voting threshold is reached.
-	Certificate *agreement.Certificate
+	Certificate *agreement.Certificate `json:"cert,omitempty"`
 }
 
 // MakeBlockDataFromValidatedBlock makes BlockData from agreement.ValidatedBlock

exporters/filewriter/file_exporter.go

@@ -13,19 +13,23 @@ import (
 
 	"github.com/algorand/go-algorand/crypto"
 	"github.com/algorand/go-algorand/data/bookkeeping"
-
 	"github.com/algorand/indexer/data"
 	"github.com/algorand/indexer/exporters"
 	"github.com/algorand/indexer/plugins"
+	"github.com/algorand/indexer/util"
 )
 
-const exporterName = "file_writer"
+const (
+	exporterName = "file_writer"
+	// FilePattern is used to name the output files.
+	FilePattern = "%[1]d_block.json"
+)
 
 type fileExporter struct {
 	round             uint64
 	blockMetadataFile string
 	blockMetadata     BlockMetaData
-	cfg               ExporterConfig
+	cfg               Config
 	logger            *logrus.Logger
 }
 
@@ -58,11 +62,16 @@ func (exp *fileExporter) Metadata() exporters.ExporterMetadata {
 
 func (exp *fileExporter) Init(_ context.Context, cfg plugins.PluginConfig, logger *logrus.Logger) error {
 	exp.logger = logger
-	if err := exp.unmarhshalConfig(string(cfg)); err != nil {
+	var err error
+	exp.cfg, err = unmarshalConfig(string(cfg))
+	if err != nil {
 		return fmt.Errorf("connect failure in unmarshalConfig: %w", err)
 	}
+	if exp.cfg.FilenamePattern == "" {
+		exp.cfg.FilenamePattern = FilePattern
+	}
 	// create block directory
-	err := os.Mkdir(exp.cfg.BlocksDir, 0755)
+	err = os.Mkdir(exp.cfg.BlocksDir, 0755)
 	if err != nil && errors.Is(err, os.ErrExist) {
 	} else if err != nil {
 		return fmt.Errorf("Init() error: %w", err)
@@ -113,12 +122,7 @@ func (exp *fileExporter) Config() plugins.PluginConfig {
 
 func (exp *fileExporter) encodeMetadataToFile() error {
 	tempFilename := fmt.Sprintf("%s.temp", exp.blockMetadataFile)
-	file, err := os.Create(tempFilename)
-	if err != nil {
-		return fmt.Errorf("encodeMetadataToFile(): failed to create temp metadata file: %w", err)
-	}
-	defer file.Close()
-	err = json.NewEncoder(file).Encode(exp.blockMetadata)
+	err := util.EncodeToFile(tempFilename, exp.blockMetadata, true)
 	if err != nil {
 		return fmt.Errorf("encodeMetadataToFile(): failed to write temp metadata: %w", err)
 	}
@@ -143,21 +147,24 @@ func (exp *fileExporter) Receive(exportData data.BlockData) error {
 	if exportData.Round() != exp.round {
 		return fmt.Errorf("Receive(): wrong block: received round %d, expected round %d", exportData.Round(), exp.round)
 	}
-	//write block to file
-	blockFile := path.Join(exp.cfg.BlocksDir, fmt.Sprintf("block_%d.json", exportData.Round()))
-	file, err := os.OpenFile(blockFile, os.O_WRONLY|os.O_CREATE, 0755)
-	if err != nil {
-		return fmt.Errorf("Receive(): error opening file %s: %w", blockFile, err)
-	}
-	defer file.Close()
-	err = json.NewEncoder(file).Encode(exportData)
-	if err != nil {
-		return fmt.Errorf("Receive(): error encoding exportData: %w", err)
+
+	// write block to file
+	{
+		if exp.cfg.DropCertificate {
+			exportData.Certificate = nil
+		}
+
+		blockFile := path.Join(exp.cfg.BlocksDir, fmt.Sprintf(exp.cfg.FilenamePattern, exportData.Round()))
+		err := util.EncodeToFile(blockFile, exportData, true)
+		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)
 	}
-	exp.logger.Infof("Added block %d", exportData.Round())
+
 	exp.round++
 	exp.blockMetadata.NextRound = exp.round
-	err = exp.encodeMetadataToFile()
+	err := exp.encodeMetadataToFile()
 	if err != nil {
 		return fmt.Errorf("Receive() metadata encoding err %w", err)
 	}
@@ -168,12 +175,17 @@ func (exp *fileExporter) HandleGenesis(genesis bookkeeping.Genesis) error {
 	// check genesis hash
 	gh := crypto.HashObj(genesis).String()
 	if exp.blockMetadata.GenesisHash == "" {
+		// First time initialization.
 		exp.blockMetadata.GenesisHash = gh
 		exp.blockMetadata.Network = string(genesis.Network)
-		err := exp.encodeMetadataToFile()
-		if err != nil {
+		if err := exp.encodeMetadataToFile(); err != nil {
 			return fmt.Errorf("HandleGenesis() metadata encoding err %w", err)
 		}
+
+		genesisFilename := path.Join(exp.cfg.BlocksDir, "genesis.json")
+		if err := util.EncodeToFile(genesisFilename, genesis, true); err != nil {
+			return fmt.Errorf("HandleGenesis() failed to serialize genesis file: %w", err)
+		}
 	} else {
 		if exp.blockMetadata.GenesisHash != gh {
 			return fmt.Errorf("HandleGenesis() genesis hash in metadata does not match expected value: actual %s, expected %s", gh, exp.blockMetadata.GenesisHash)
@@ -186,8 +198,10 @@ func (exp *fileExporter) Round() uint64 {
 	return exp.round
 }
 
-func (exp *fileExporter) unmarhshalConfig(cfg string) error {
-	return yaml.Unmarshal([]byte(cfg), &exp.cfg)
+func unmarshalConfig(cfg string) (Config, error) {
+	var config Config
+	err := yaml.Unmarshal([]byte(cfg), &config)
+	return config, err
 }
 
 func init() {

exporters/filewriter/file_exporter_config.go

@@ -1,9 +1,21 @@
 package filewriter
 
-// ExporterConfig specific to the file exporter
-type ExporterConfig struct {
-	// full file path to a directory
-	// where the block data should be stored.
-	// Create if directory doesn't exist
+// Config specific to the file exporter
+type Config struct {
+	// BlocksDir is the path to a directory where block data should be stored.
+	// The directory is created if it doesn't exist.
 	BlocksDir string `yaml:"block-dir"`
+	// 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"
+	FilenamePattern string `yaml:"filename-pattern"`
+	// DropCertificate is used to remove the vote certificate from the block data before writing files.
+	DropCertificate bool `yaml:"drop-certificate"`
+
+	// TODO: compression level - Default, Fastest, Best compression, etc
+
+	// TODO: How to avoid having millions of files in a directory?
+	//       Write batches of blocks to a single file?
+	//       Tree of directories
 }