Function sourcesFromFile

pkg/acquisition/acquisition.go:319–390

sourcesFromFile reads and parses one acquisition file into DataSources.

(
	ctx context.Context,
	acquisFile string,
	metricsLevel metrics.AcquisitionMetricsLevel,
	hub *cwhub.Hub,
)

// sourcesFromFile reads and parses one acquisition file into DataSources.
func sourcesFromFile(
	ctx context.Context,
	acquisFile string,
	metricsLevel metrics.AcquisitionMetricsLevel,
	hub *cwhub.Hub,
) ([]types.DataSource, error) {
	var sources []types.DataSource

	log.Infof("loading acquisition file : %s", acquisFile)

	yamlFile, err := os.Open(acquisFile)
	if err != nil {
		return nil, err
	}

	defer yamlFile.Close()

	acquisContent, err := io.ReadAll(yamlFile)
	if err != nil {
		return nil, fmt.Errorf("failed to read %s: %w", acquisFile, err)
	}

	expandedAcquis := csstring.StrictExpand(string(acquisContent), os.LookupEnv)

	documents, err := csyaml.SplitDocuments(strings.NewReader(expandedAcquis))
	if err != nil {
		return nil, err
	}

	idx := -1

	for _, yamlDoc := range documents {
		idx += 1

		loc := formatConfigLocation(acquisFile, len(documents) > 1, idx)

		parsed, err := ParseSourceConfig(ctx, yamlDoc, metricsLevel, hub)

		// report data source detection, it can be required to understand an error
		if parsed != nil {
			if parsed.SourceMissing {
				log.Debugf("%s: datasource type missing, detected 'source=%s'", loc, parsed.Common.Source)
			}

			if parsed.SourceOverridden != "" {
				log.Warnf("%s: datasource type mismatch: found '%s' but should probably be '%s'", loc, parsed.SourceOverridden, parsed.Common.Source)
			}
		}

		if err != nil {
			if errors.Is(err, ErrEmptyYAMLDocument) {
				continue
			}

			var dserr *DataSourceUnavailableError
			if errors.As(err, &dserr) {
				log.Error(fmt.Errorf("%s: %w", loc, err))
				continue