func runAgent()

in cmd/amazon-cloudwatch-agent/amazon-cloudwatch-agent.go [231:396]
128 lines of code
29 McCabe index (conditional complexity)

func runAgent(ctx context.Context,
	inputFilters []string,
	outputFilters []string,
) error {
	envConfigPath, err := getEnvConfigPath(*fTomlConfig, *fEnvConfig)
	if err != nil {
		return err
	}
	err = loadEnvironmentVariables(envConfigPath)
	if err != nil && !*fSchemaTest {
		log.Printf("W! Failed to load environment variables due to %s\n", err.Error())
	}
	// If no other options are specified, load the config file and run.
	c := config.NewConfig()
	c.OutputFilters = outputFilters
	c.InputFilters = inputFilters
	c.AllowUnusedFields = true

	err = loadTomlConfigIntoAgent(c)
	if err != nil {
		return err
	}

	err = validateAgentFinalConfigAndPlugins(c)
	if err != nil {
		return err
	}

	ag, err := agent.NewAgent(c)
	if err != nil {
		return err
	}

	// Setup logging as configured.
	logConfig := logger.LogConfig{
		Debug:               ag.Config.Agent.Debug || *fDebug,
		Quiet:               ag.Config.Agent.Quiet || *fQuiet,
		LogTarget:           ag.Config.Agent.LogTarget,
		Logfile:             ag.Config.Agent.Logfile,
		RotationInterval:    ag.Config.Agent.LogfileRotationInterval,
		RotationMaxSize:     ag.Config.Agent.LogfileRotationMaxSize,
		RotationMaxArchives: ag.Config.Agent.LogfileRotationMaxArchives,
		LogWithTimezone:     "",
	}

	writer := logger.NewLogWriter(logConfig)

	log.Printf("I! Starting AmazonCloudWatchAgent %s with log file %s with log target %s\n", version.Full(), ag.Config.Agent.Logfile, ag.Config.Agent.LogTarget)
	// Need to set SDK log level before plugins get loaded.
	// Some aws.Config objects get created early and live forever which means
	// we cannot change the sdk log level without restarting the Agent.
	// For example CloudWatch.Connect().
	sdkLogLevel := os.Getenv(envconfig.AWS_SDK_LOG_LEVEL)
	configaws.SetSDKLogLevel(sdkLogLevel)
	if sdkLogLevel == "" {
		log.Println("I! AWS SDK log level not set")
	} else {
		log.Printf("I! AWS SDK log level, %s\n", sdkLogLevel)
	}

	if *fTest || *fTestWait != 0 {
		testWaitDuration := time.Duration(*fTestWait) * time.Second
		return ag.Test(ctx, testWaitDuration)
	}
	if *fPidfile != "" {
		f, err := os.OpenFile(*fPidfile, os.O_CREATE|os.O_WRONLY, 0644)
		if err != nil {
			log.Printf("E! Unable to create pidfile: %s", err)
		} else {
			fmt.Fprintf(f, "%d\n", os.Getpid())

			f.Close()

			defer func() {
				err := os.Remove(*fPidfile)
				if err != nil {
					log.Printf("E! Unable to remove pidfile: %s\n", err)
				}
			}()
		}
	}
	if envconfig.IsRunningInROSA() {
		log.Println("I! Running in ROSA")
	}

	if envconfig.IsSelinuxEnabled() {
		log.Println("I! SELinux Status: Enabled")
	}

	if len(c.Inputs) != 0 && len(c.Outputs) != 0 {
		log.Println("creating new logs agent")
		logAgent := logs.NewLogAgent(c)
		// Always run logAgent as goroutine regardless of whether starting OTEL or Telegraf.
		go logAgent.Run(ctx)

		// If only a single YAML is provided and does not exist, then ASSUME the agent is
		// just monitoring logs since this is the default when no OTEL config flag is provided.
		// So just start Telegraf.
		if len(fOtelConfigs) == 1 {
			_, err = os.Stat(fOtelConfigs[0])
			if errors.Is(err, os.ErrNotExist) {
				log.Println("I! running in logs-only mode")
				useragent.Get().SetComponents(&otelcol.Config{}, c)
				return ag.Run(ctx)
			}
		}
	}
	// Else start OTEL and rely on adapter package to start the logfile plugin.
	level := cwaLogger.ConvertToAtomicLevel(wlog.LogLevel())
	logger, loggerOptions := cwaLogger.NewLogger(writer, level)

	otelConfigs := fOtelConfigs
	// try merging configs together, will return nil if nothing to merge
	merged, err := mergeConfigs(otelConfigs)
	if err != nil {
		return err
	}
	if merged != nil {
		_ = os.Setenv(envconfig.CWAgentMergedOtelConfig, toyamlconfig.ToYamlConfig(merged.ToStringMap()))
		otelConfigs = []string{"env:" + envconfig.CWAgentMergedOtelConfig}
	} else {
		_ = os.Unsetenv(envconfig.CWAgentMergedOtelConfig)
	}

	providerSettings := configprovider.GetSettings(otelConfigs, logger)
	provider, err := otelcol.NewConfigProvider(providerSettings)
	if err != nil {
		return fmt.Errorf("error while initializing config provider: %v", err)
	}

	factories, err := components(c)
	if err != nil {
		return fmt.Errorf("error while adapting telegraf input plugins: %v", err)
	}

	cfg, err := provider.Get(ctx, factories)
	if err != nil {
		return err
	}

	if _, ok := os.LookupEnv(envconfig.CWAgentMergedOtelConfig); ok {
		result, err := mapstructure.Marshal(cfg)
		if err != nil {
			return fmt.Errorf("failed to marshal OTEL configuration: %v", err)
		}
		log.Printf("I! Merged OTEL configuration: \n%s\n", toyamlconfig.ToYamlConfig(result))
	}

	useragent.Get().SetComponents(cfg, c)

	params := getCollectorParams(factories, providerSettings, loggerOptions)
	cmd := otelcol.NewCommand(params)
	// *************************************************************************************************
	// ⚠️ WARNING ⚠️
	// Noticed that args of parent process get passed here to otel collector which causes failures
	// complaining about unrecognized args. So below change overwrites the args.
	// The config path below here is actually used that was set in the settings above.
	// docs: https://github.com/open-telemetry/opentelemetry-collector/blob/93cbae436ae61b832279dbbb18a0d99214b7d305/otelcol/command.go#L63
	// *************************************************************************************************
	var e []string
	for _, uri := range otelConfigs {
		e = append(e, "--config="+uri)
	}
	cmd.SetArgs(e)
	return cmd.Execute()
}