Method CatLogStream

pkg/acquisition/modules/cloudwatch/run.go:378–446

(ctx context.Context, cfg *LogStreamTailConfig, outChan chan pipeline.Event)

}

func (s *Source) CatLogStream(ctx context.Context, cfg *LogStreamTailConfig, outChan chan pipeline.Event) error {
	var startFrom *string

	head := true
	// convert the times
	startTime := cfg.StartTime.UTC().Unix() * 1000
	endTime := cfg.EndTime.UTC().Unix() * 1000

	hasMoreEvents := true
	for hasMoreEvents {
		select {
		default:
			cfg.logger.Tracef("Calling GetLogEventsPagesWithContext(%s, %s), startTime:%d / endTime:%d",
				cfg.GroupName, cfg.StreamName, startTime, endTime)
			cfg.logger.Tracef("startTime:%s / endTime:%s", cfg.StartTime, cfg.EndTime)

			if startFrom != nil {
				cfg.logger.Tracef("next_token: %s", *startFrom)
			}

			p := cloudwatchlogs.NewGetLogEventsPaginator(
				s.cwClient,
				&cloudwatchlogs.GetLogEventsInput{
					Limit:         aws.Int32(10),
					LogGroupName:  aws.String(cfg.GroupName),
					LogStreamName: aws.String(cfg.StreamName),
					StartTime:     aws.Int64(startTime),
					EndTime:       aws.Int64(endTime),
					StartFromHead: &head,
					NextToken:     startFrom,
				},
				)
			for p.HasMorePages() {
				page, err := p.NextPage(ctx)
				if err != nil {
					return fmt.Errorf("while reading logs from %s/%s: %w", cfg.GroupName, cfg.StreamName, err)
				}

				for _, e := range page.Events {
					evt, err := cwLogToEvent(e, cfg)
					if err != nil {
						cfg.logger.Warningf("discard event: %s", err)
					}

					cfg.logger.Debugf("pushing message: %s", evt.Line.Raw)

					outChan <- evt
				}

				if startFrom != nil && page.NextForwardToken != nil && *page.NextForwardToken == *startFrom {
					cfg.logger.Debugf("reached end of available events")
					hasMoreEvents = false
					break
				}

				startFrom = page.NextForwardToken
			}