NewForFlusher constructs a new Ingester to be used by flusher target. Compared to the 'New' method: - Always replays the WAL. - Does not start the lifecycler. this is a special version of ingester used by Flusher. This ingester is not ingesting anything, its only purpose is to react on Flush method
(cfg Config, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger)
| 891 | // this is a special version of ingester used by Flusher. This ingester is not ingesting anything, its only purpose is to react |
| 892 | // on Flush method and flush all opened TSDBs when called. |
| 893 | func NewForFlusher(cfg Config, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error) { |
| 894 | bucketClient, err := bucket.NewClient(context.Background(), cfg.BlocksStorageConfig.Bucket, nil, "ingester", logger, registerer) |
| 895 | if err != nil { |
| 896 | return nil, errors.Wrap(err, "failed to create the bucket client") |
| 897 | } |
| 898 | |
| 899 | i := &Ingester{ |
| 900 | cfg: cfg, |
| 901 | limits: limits, |
| 902 | TSDBState: newTSDBState(bucketClient, registerer), |
| 903 | logger: logger, |
| 904 | } |
| 905 | i.limiter = NewLimiter( |
| 906 | limits, |
| 907 | i.lifecycler, |
| 908 | cfg.DistributorShardingStrategy, |
| 909 | cfg.DistributorShardByAllLabels, |
| 910 | cfg.LifecyclerConfig.RingConfig.ReplicationFactor, |
| 911 | cfg.LifecyclerConfig.RingConfig.ZoneAwarenessEnabled, |
| 912 | cfg.AdminLimitMessage, |
| 913 | ) |
| 914 | i.metrics = newIngesterMetrics(registerer, |
| 915 | false, |
| 916 | false, |
| 917 | false, |
| 918 | i.getInstanceLimits, |
| 919 | nil, |
| 920 | &i.maxInflightPushRequests, |
| 921 | &i.maxInflightQueryRequests, |
| 922 | cfg.BlocksStorageConfig.TSDB.PostingsCache.Blocks.Enabled || cfg.BlocksStorageConfig.TSDB.PostingsCache.Head.Enabled, |
| 923 | cfg.EnableRegexMatcherLimits, |
| 924 | ) |
| 925 | |
| 926 | i.TSDBState.shipperIngesterID = "flusher" |
| 927 | |
| 928 | // This ingester will not start any subservices (lifecycler, compaction, shipping), |
| 929 | // and will only open TSDBs, wait for Flush to be called, and then close TSDBs again. |
| 930 | i.BasicService = services.NewIdleService(i.startingV2ForFlusher, i.stoppingV2ForFlusher) |
| 931 | return i, nil |
| 932 | } |
| 933 | |
| 934 | func (i *Ingester) startingV2ForFlusher(ctx context.Context) error { |
| 935 | if err := i.openExistingTSDB(ctx); err != nil { |