Function StartAcquisition

pkg/acquisition/acquisition.go:614–666

(
	ctx context.Context,
	sources []types.DataSource,
	output chan pipeline.Event,
	acquisTomb *tomb.Tomb,
)

}

func StartAcquisition(
	ctx context.Context,
	sources []types.DataSource,
	output chan pipeline.Event,
	acquisTomb *tomb.Tomb,
) error {
	// Don't wait if we have no sources, as it will hang forever
	if len(sources) == 0 {
		return nil
	}

	for i := range sources {
		subsrc := sources[i] // ensure it's a copy
		log.Debugf("starting one source %d/%d ->> %T", i, len(sources), subsrc)

		acquisTomb.Go(func() error {
			defer trace.ReportPanic()

			outChan := output

			log.Debugf("datasource %s UUID: %s", subsrc.GetName(), subsrc.GetUuid())

			if transformRuntime, ok := transformRuntimes[subsrc.GetUuid()]; ok {
				log.Infof("transform expression found for datasource %s", subsrc.GetName())

				transformChan := make(chan pipeline.Event)
				outChan = transformChan
				transformLogger := log.WithFields(log.Fields{
					"component":  "transform",
					"datasource": subsrc.GetName(),
				})

				acquisTomb.Go(func() error {
					defer trace.ReportPanic()
					transform(outChan, output, acquisTomb, transformRuntime, transformLogger)
					return nil
				})
			}

			if err := acquireSource(ctx, subsrc, subsrc.GetName(), output, acquisTomb); err != nil {
				// if one of the acquisitions returns an error, we kill the others to properly shutdown
				acquisTomb.Kill(err)
			}

			return nil
		})
	}

	// return only when acquisition is over (cat) or never (tail)
	err := acquisTomb.Wait()

	return err
}