Function GetStreamingLogs

actor/sharedaction/logging.go:124–172

(appGUID string, client LogCacheClient)

}

func GetStreamingLogs(appGUID string, client LogCacheClient) (<-chan LogMessage, <-chan error, context.CancelFunc) {

	logrus.Info("Start Tailing Logs")

	outgoingLogStream := make(chan LogMessage, 1000)
	outgoingErrStream := make(chan error, 1000)
	ctx, cancelFunc := context.WithCancel(context.Background())
	go func() {
		defer close(outgoingLogStream)
		defer close(outgoingErrStream)

		ts := latestEnvelopeTimestamp(client, outgoingErrStream, ctx, appGUID)

		// if the context was cancelled we may not have seen an envelope
		if ts.IsZero() {
			return
		}

		const offset = 1 * time.Second
		walkStartTime := ts.Add(-offset)

		logcache.Walk(
			ctx,
			appGUID,
			logcache.Visitor(func(envelopes []*loggregator_v2.Envelope) bool {
				logMessages := convertEnvelopesToLogMessages(envelopes)
				for _, logMessage := range logMessages {
					select {
					case <-ctx.Done():
						return false
					default:
						outgoingLogStream <- *logMessage
					}
				}
				return true
			}),
			client.Read,
			logcache.WithWalkDelay(2*time.Second),
			logcache.WithWalkStartTime(walkStartTime),
			logcache.WithWalkEnvelopeTypes(logcache_v1.EnvelopeType_LOG),
			logcache.WithWalkBackoff(newCliRetryBackoff(retryInterval, retryCount)),
			logcache.WithWalkLogger(log.New(channelWriter{
				errChannel: outgoingErrStream,
			}, "", 0)),
		)
	}()

	return outgoingLogStream, outgoingErrStream, cancelFunc
}

func latestEnvelopeTimestamp(client LogCacheClient, errs chan error, ctx context.Context, sourceID string) time.Time {