typos

conduit/pipeline/pipeline.go

@@ -108,8 +108,8 @@
 func (p *pipelineImpl) makeConfig(cfg data.NameConfigPair, pluginType plugins.PluginType) (*log.Logger, plugins.PluginConfig, error) {
 	configs, err := yaml.Marshal(cfg.Config)
 	if err != nil {
 		return nil, plugins.PluginConfig{}, fmt.Errorf("makeConfig(): could not serialize config: %w", err)
 	}
 
 	lgr := log.New()
 	lgr.SetOutput(p.logger.Out)
@@ -122,8 +122,8 @@
 		config.DataDir = path.Join(p.cfg.ConduitArgs.ConduitDataDir, fmt.Sprintf("%s_%s", pluginType, cfg.Name))
 		err = os.MkdirAll(config.DataDir, os.ModePerm)
 		if err != nil {
 			return nil, plugins.PluginConfig{}, fmt.Errorf("makeConfig: unable to create plugin data directory: %w", err)
 		}
 	}
 
 	return lgr, config, nil
@@ -173,8 +173,8 @@
 	for _, part := range parts {
 		_, config, err := p.makeConfig(part.cfg, part.t)
 		if err != nil {
 			return 0, err
 		}
 		rnd, err := part.RoundRequest(config)
 		if err != nil {
 			return 0, err
@@ -204,16 +204,16 @@
 	telemetryConfig := telemetry.MakeTelemetryConfig(p.cfg.Telemetry.URI, p.cfg.Telemetry.Index, p.cfg.Telemetry.UserName, p.cfg.Telemetry.Password)
 	telemetryClient, err := telemetry.MakeOpenSearchClient(telemetryConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize telemetry: %w", err)
 	}
 	p.logger.Infof("Telemetry initialized with URI: %s", telemetryConfig.URI)
 
 	// If GUID is not in metadata, save it. Otherwise, use the GUID from metadata.
 	if p.pipelineMetadata.TelemetryID == "" {
 		p.pipelineMetadata.TelemetryID = telemetryClient.TelemetryConfig.GUID
 	} else {
 		telemetryClient.TelemetryConfig.GUID = p.pipelineMetadata.TelemetryID
 	}
 
 	return telemetryClient, nil
 }
@@ -232,12 +232,12 @@
 		var err error
 		profFile, err := os.Create(p.cfg.CPUProfile)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): unable to create profile: %w", err)
 		}
 		p.profFile = profFile
 		err = pprof.StartCPUProfile(profFile)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): unable to start pprof: %w", err)
 		}
 	}
 
@@ -286,7 +286,7 @@
 		var telemetryErr error
 		telemetryClient, telemetryErr = p.initializeTelemetry()
 		if telemetryErr != nil {
 			p.logger.Warnf("Telemetry initialization failed, continuing without telemetry: %s", telemetryErr)
 		} else {
 			// Try sending a startup event. If it fails, log a warning and continue
 			event := telemetryClient.MakeTelemetryStartupEvent()
@@ -304,16 +304,16 @@
 	{
 		importerLogger, pluginConfig, err := p.makeConfig(p.cfg.Importer, plugins.Importer)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not make %s config: %w", p.cfg.Importer.Name, err)
 		}
 		err = (*p.importer).Init(p.ctx, *p.initProvider, pluginConfig, importerLogger)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize importer (%s): %w", p.cfg.Importer.Name, err)
 		}
 		genesis, err := (*p.importer).GetGenesis()
 		if err != nil {
 			return fmt.Errorf("Pipeline.GetGenesis(): could not obtain Genesis from the importer (%s): %w", p.cfg.Importer.Name, err)
 		}
 		(*p.initProvider).SetGenesis(genesis)
 
 		// write pipeline metadata
@@ -326,8 +326,8 @@
 		p.pipelineMetadata.Network = genesis.Network
 		err = p.pipelineMetadata.encodeToFile(p.cfg.ConduitArgs.ConduitDataDir)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init() failed to write metadata to file: %w", err)
 		}
 
 		p.logger.Infof("Initialized Importer: %s", p.cfg.Importer.Name)
 	}
@@ -337,11 +337,11 @@
 		ncPair := p.cfg.Processors[idx]
 		logger, config, err := p.makeConfig(ncPair, plugins.Processor)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize processor (%s): %w", ncPair, err)
 		}
 		err = (*processor).Init(p.ctx, *p.initProvider, config, logger)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize processor (%s): %w", ncPair.Name, err)
 		}
 		p.logger.Infof("Initialized Processor: %s", ncPair.Name)
 	}
@@ -350,12 +350,12 @@
 	{
 		logger, config, err := p.makeConfig(p.cfg.Exporter, plugins.Exporter)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize processor (%s): %w", p.cfg.Exporter.Name, err)
 		}
 		err = (*p.exporter).Init(p.ctx, *p.initProvider, config, logger)
 		if err != nil {
-			return fmt.Errorf("Pipeline.Start(): could not initialize Exporter (%s): %w", p.cfg.Exporter.Name, err)
+			return fmt.Errorf("Pipeline.Init(): could not initialize Exporter (%s): %w", p.cfg.Exporter.Name, err)
 		}
 		p.logger.Infof("Initialized Exporter: %s", p.cfg.Exporter.Name)
 	}
 
@@ -615,22 +615,22 @@
 	logger.Infof("Creating PID file at: %s", pidFilePath)
 	fout, err := os.Create(pidFilePath)
 	if err != nil {
 		err = fmt.Errorf("%s: could not create pid file, %v", pidFilePath, err)
 		logger.Error(err)
 		return err
 	}
 
 	if _, err = fmt.Fprintf(fout, "%d", os.Getpid()); err != nil {
 		err = fmt.Errorf("%s: could not write pid file, %v", pidFilePath, err)
 		logger.Error(err)
 		return err
 	}
 
 	err = fout.Close()
 	if err != nil {
 		err = fmt.Errorf("%s: could not close pid file, %v", pidFilePath, err)
 		logger.Error(err)
 		return err
 	}
 	return err
 }

docs/Development.md

@@ -56,7 +56,7 @@ Generally speaking, you can follow the code in one of the existing plugins.
 
 ### Init
 
-Each plugin will have its `Init` function called once as the pipline is constructed.
+Each plugin will have its `Init` function called once as the pipeline is constructed.
 
 The context provided to this function should be saved, and used to terminate any long-running operations if necessary.