Concept for patch: source startup delay

[johnhtodd]

We now are using memory enrichment tables extensively - they're really useful, and are I think are a not-well-understood function of Vector that has amazing functionality. We're also using the sink-based storage function for memory tables: periodically storing the contents of those memory enrichment tables to disk. Why? Because we re-load them when Vector restarts, so we don't have to generate the whole table again. Basically: we're saving state of how to deal with or modify objects passing through Vector, so if a system crashes (or comes up as "new") we can create the memory table as it was with some reasonable belief that it is close to what it was when the system went away.

But there is a problem: loading these disk files of saved state information into the memory tables takes time. Sometimes a few seconds, and in the worst cases, it takes many minutes (we have some tables with hundreds of millions of records.) But our other sources (files or kafka sources in our case) start up immediately, and so we have objects flowing through the Vector system, but we have incomplete ability to apply the correct actions to those objects because the memory tables aren't full yet. So we're processing events with "rules" that aren't complete. This leads to very unpredictable outcomes that are often wildly incorrect.

Currently, we use some really ugly hacks to solve this, such as blocking kafka origins with packet filters, or causing other transient faults that we can manually clear when the state tables are ready for the other sources to start feeding data through the system.

My memory table example is just one reason that one might want to delay or halt ingestion of a source. I expect others have different but equally valid reasons that some sources should be delayed or ignored.

I would propose some method by which sources could be delayed upon startup, so sources could be activated in an orderly, well-understood sequence rather than all at once. This would apply to all sources in their general configuration parameter section.

The most trivial way of doing this would be a seconds-based timer that simply waited until expiry to begin processing the source.

A much more flexible way to do this would be the ability to run some small VRL script on a regular basis (specifiable interval in the source config?) that would result in a "true" or "false" outcome. The source would be enabled or disabled based on this outcome. We use a variant of this type of logic already in the user authentication model in the websocket sink (and other places?) where standalone VRL is used as a method to generate a true/false outcome. Once we have the internal metrics visible via VRL (this PR: #23430) then it would be easy to examine the uptime of the vector itself and emulate the "delay for X seconds" model that I suggest as the "trivial" solution above. But a programmatic method would open the door to much more sophisticated methods of evaluating and activating/deactivating sources. This perhaps is parallel to the "backpressure" model that is already used, but it would also be much simpler and would not have buffering issues. It is not a replacement for backpressure, but would be an adjunct method. Sources would just be turned off until the VRL was "true".

I'd like to get any opinions on which of these methods makes sense, or perhaps neither, or perhaps there other models that already exist but I haven't thought of which are a better solution.

[pront]

Hi @johnhtodd, I would recommend writing an RFC before we go ahead with any implementation.

But a programmatic method would open the door to much more sophisticated methods of evaluating and activating/deactivating sources.

This is very interesting and we don't offer any flexibility in this area. Today, we build all or nothing. I could imagine a new synchronization mechanism where we tell Vector to wait until certain conditions are met but there are several questions here:

• how can users define conditions? VRL programs?
• how do we observe these conditions? is it automatic or user sends a signal to the Vector process?
• to evaluate the condition, do we need to run Vector to get internal telemetry and/or Vector components?
  • This implies that we have to run a pipeline before and then another pipeline after. Could we have a Vector (prepare pipeline) to Vector (actual pipeline) setup?

[johnhtodd]

Do you want to run a Vector pipeline to pass events to the VRL conditions? Or what data will the VRL conditions depend on? Purely outside of Vector and available at startup?

I don't think it makes sense to pass events; that seems overly complex and sort of not consistent with the way it's been done before. If there is the ability to connect to and observe things on memory enrichment tables, and also to access vector metrics (per the PR that is pending) then for me that seems sufficient.

[pront]

I don't think it makes sense to pass events; that seems overly complex and sort of not consistent with the way it's been done before.

We are discussing a new feature so not sure what you mean by "done before" in this context.

If there is the ability to connect to and observe things on memory enrichment tables, and also to access vector metrics (per the PR that is pending) then for me that seems sufficient.

Let's wait for the RFC and discuss the details there. I expect some problems will become more obvious as we dive deeper.

[johnhtodd]

I don't think it makes sense to pass events; that seems overly complex and sort of not consistent with the way it's been done before.

We are discussing a new feature so not sure what you mean by "done before" in this context.

Sorry - wasn't clear. I meant that VRL in a semi-isolated method (not taking steady stream of events) to output true/false decisions has already been done in the websocket authentication model.

If there is the ability to connect to and observe things on memory enrichment tables, and also to access vector metrics (per the PR that is pending) then for me that seems sufficient.

Let's wait for the RFC and discuss the details there. I expect some problems will become more obvious as we dive deeper.

OK. It will take me a while on this.

[watson28]

Hi there, was there any update on the RFC? I find myself in the need for this feature in vector: pretty much the same use case as @johnhtodd.

[watson28]

related to this topic ... yesterday there was a new vector release that includes a delay transform: https://github.com/vectordotdev/vector/releases/tag/v0.56.0

[johnhtodd]

I suppose you could use the "delay" transform that did some math against an environment variable to wait until the appropriate clock time has passed. To answer your question directly: To my knowledge, no work was done on a method to delay as an absolute. I've not checked but I suspect the delay method will overflow buffers and drop events if you delay for too long (that might be worth validating!) so I'm not sure that method will work without data loss for the case I mentioned at the top of this ticket.

[pront]

Hello, sorry for the late response. I was going over unanswered discussions and I saw this thread. Is the delay transform the answer to this discussion? If not, we can continue here.

[johnhtodd]

Hello, sorry for the late response. I was going over unanswered discussions and I saw this thread. Is the delay transform the answer to this discussion? If not, we can continue here.

No, the "delay" transform doesn't work as that's quite challenging to make work in a "startup only" mode. It can be sort-of done with delay transforms with booleans based on number of seconds of uptime, but that also then runs into an immediate problem with 100% full buffers. It's a really ugly hack.

I'd still prefer to have sources delayed by a certain number of seconds after startup - this is the easiest way to proceed with this functionality.

A "bonus extra credit" method for this would be to use a boolean VRL function to be able to activate/de-active a source repeatedly, based on any arbitrary criteria that could be expressed in VRL. Then each source could have an optional "active" parameter which could then take a short VRL evaluation to determine if the source should be active or inactive. It would have to be evaluated every N seconds (not sure what "N" should be here) but that would allow some granularity on turning on sources based on very customizable criteria.

Example, storing the "uptime_seconds" value from the internal metrics into a memory enrichment table, and then making that available to VRL (haven't tried this, but I think? it should work):

[sources.internal]
type = "internal_metrics"
scrape_interval_secs = 10

[transforms.uptime_only]
type = "filter"
inputs = ["internal"]
condition = '.name == "uptime_seconds"'

[transforms.uptime_log]
type = "metric_to_log"
inputs = ["uptime_only"]

[transforms.uptime_kv]
type = "remap"
inputs = ["uptime_log"]
source = '''
. = { "uptime": { "seconds": to_int!(.gauge.value) } }
'''

[enrichment_tables.uptime_table]
type = "memory"
inputs = ["uptime_kv"]
ttl = 60
flush_interval = 60

then we end up with a boolean like this where 'should_i_consume' is true only if uptime is more than 60 seconds:

should_i_consume = get_enrichment_table_record!("uptime_table", { "key": "uptime" }).value.seconds > 60