add maxlag

config.go

@@ -9,20 +9,26 @@ import (
 	ltsregion "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/lts/v2/region"
 )
 
-var maxEndFromNow = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MAX_END_FROM_NOW"), time.Minute)
-var maxFetchRange = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MAX_FETCH_RANGE"), 5*time.Minute)
-var minFetchRange = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MIN_FETCH_RANGE"), time.Minute)
-var streamRoutine = int(strToIntOrDefault(os.Getenv("HWC_LOGSTREAM_ROUTINE"), 5))
-var streamPosTag = envOrDefault("HWC_LOGSTREAM_POSITITION_TAG", "x-hwc-logstream-pos")
-var streamExclusionTag = envOrDefault("HWC_LOGSTREAM_EXCLUSION_TAG", "x-hwc-logstream-exclude")
-var regionID = envOrDefault("HUAWEICLOUD_SDK_REGION_ID", "ap-southeast-4")
-
-var addRegion = map[string]*region.Region{
+var (
+	maxEndFromNow = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MAX_END_FROM_NOW"), time.Minute)
+	maxFetchRange = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MAX_FETCH_RANGE"), 5*time.Minute)
+	minFetchRange = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MIN_FETCH_RANGE"), time.Minute)
+	maxLag        = strToDurationOrDefault(os.Getenv("HWC_LOGSTREAM_MAX_LAG"), 24*time.Hour)
+
+	streamRoutine = int(strToIntOrDefault(os.Getenv("HWC_LOGSTREAM_ROUTINE"), 5))
+
+	streamPosTag       = envOrDefault("HWC_LOGSTREAM_POSITITION_TAG", "x-hwc-logstream-pos")
+	streamExclusionTag = envOrDefault("HWC_LOGSTREAM_EXCLUSION_TAG", "x-hwc-logstream-exclude")
+
+	regionID = envOrDefault("HUAWEICLOUD_SDK_REGION_ID", "ap-southeast-4")
+)
+
+var additionalRegion = map[string]*region.Region{
 	"ap-southeast-4": region.NewRegion("ap-southeast-4", "https://lts.ap-southeast-4.myhuaweicloud.com"),
 }
 
 func regionFromEnv() *region.Region {
-	if r, ok := addRegion[regionID]; ok {
+	if r, ok := additionalRegion[regionID]; ok {
 		return r
 	}
 

main.go

@@ -34,6 +34,7 @@ func main() {
 			maxEndFromNow: maxEndFromNow,
 			maxFetchRange: maxFetchRange,
 			minFetchRange: minFetchRange,
+			maxLag:        maxLag,
 		}
 		errc <- mgr.Start(ctx, streamRoutine)
 	}()

manager.go

@@ -25,6 +25,7 @@ type LogstreamManager struct {
 	maxEndFromNow time.Duration
 	maxFetchRange time.Duration
 	minFetchRange time.Duration
+	maxLag        time.Duration
 }
 
 func (m LogstreamManager) Start(ctx context.Context, routine int) (err error) {
@@ -69,6 +70,8 @@ func (m LogstreamManager) Start(ctx context.Context, routine int) (err error) {
 					continue
 				}
 
+				s.SkiptoCatchUp(m.maxLag)
+
 				end := time.Now().Add(-m.maxEndFromNow)
 				if e := s.start.Add(m.maxFetchRange); e.Before(end) {
 					end = e

stream.go

@@ -222,3 +222,11 @@ func (s *Logstream) SavePositition() (err error) {
 	s.start = s.last
 	return
 }
+
+func (s *Logstream) SkiptoCatchUp(maxlag time.Duration) {
+	now := time.Now()
+	if now.Sub(s.start) <= maxlag {
+		return
+	}
+	s.start = now.Add(-maxlag)
+}