File read log interval

[xufeixianggithub]

          Hi @xufeixianggithub ,

I see the issue you are describing, but I don't think we'd solve it by adding configuration to the file source. Instead, I think a more holistic way to solve this problem is for the throttle transform to support applying back-pressure. This is being tracked by #13651

You could also consider configuring the sink to apply back-pressure by limiting the concurrency or batch sizes.

I'll close this issue, but let me know if you disagree with my assessment!

Originally posted by @jszwedko in #22095 (comment)

[xufeixianggithub]

I will describe my usage scenario in detail. source configuates file reading, sink configuates elasticsearch, configuates buffering and backpressing hard buffer and block modes, but in fact, according to my test results, even if buffer files are generated, they will be used to write buffer files. However, the downstream elasticsearch cpu is still high, and the upstream source is still using 40% of the total amount on a 4-core 8G server because I configured it with only single threads. I used vector top to view, and writes to source and transforms reached tens of Kelvin per second. Downstream elasticsearch cpu usage is up to 90 percent. If I configure throttle, I have to throw out logs, and in fact, if I can slow this down without introducing new middleware like kafka, I think it's perfect.

[jszwedko]

I like the workaround you identified with Lua though. I hadn't thought of that 👍

[xufeixianggithub]

I used the way of backpressure, but the results of the test could not meet the expectations. As I said, source collection consumes a lot of cpu even if I set a single thread and write elasticsearch at the same time, the downstream is also very high cpu consumption. If the block set of backpressure can be effective, then it is also a very good way. Because he didn't live up to my expectations, so I tried to use other ways to deal with it. I know that perhaps the common solution is to pull logs from kafka, naturally you can configure the number of concurrent pull, but pull from the file is not, but the introduction of kafka is also to increase the complexity and cost of the system. What's more, I only want to solve the peak in the log reading at the beginning or one day, and I can actually read the log slowly, without the need to increase the cost of solving the peak. Can you help me see if there's a problem with my back pressure configuration,

[jszwedko]

I'm not sure I completely understand since the workaround you provided using the lua transform would also be using backpresusre as the mechanism to slow down reads. So if that works, enhancing the throttle transform to have similar backpressure behavior should work the same.

[pront]

What's more, I only want to solve the peak in the log reading at the beginning or one day, and I can actually read the log slowly, without the need to increase the cost of solving the peak.

the throttle transform to be enhanced to support blocking, rather than dropping.

^ This is what you want (currently not supported) but your Lua workaround seems to do the same thing.

Also, the comment above describes multiple different problems, ingestion and rate control.

[xufeixianggithub]

I mean, you can ignore the conversion of lua I wrote, and only look at the back pressure part configured in sink, whether there is a problem. Because I configured the backpressure, even though my capture and write es are on two different servers, the es server is still 90 percent cpu occupied, so I can't rely on the backpressure alone to solve this problem. Our daily log volume, if we can reduce the way we process data to read and write, we will not need to add new middleware and servers. And with the current server, I only need to use less than 5 percent of the cpu to solve the problem. At present, the way I use Lua, is to achieve this effect, the same test many times, the source file is 1000000 logs, and the final write is 1000000 logs. More importantly, he will not discard the log, the current official support of the traffic limiting policy, I see will discard the log, I think, since I need to collect this log, then I hope not to casually discard, if the log needs to be discarded, then I can not write to throw to the collector right.

[xufeixianggithub]

sources:
  app_logs_src:
    type: "file"
    include:
       - "/alidata/info-2025-01-01-20.log"
    ignore_older_secs: 125000     # 1 day
    ignore_checkpoints: true
    line_delimiter: "-[END]\n"
    fingerprint:
      strategy: "checksum"
      line: 1
      ignored_header_bytes: 512
    max_line_bytes: 5097152  
  internal_metrics_src:
    type: "internal_metrics"
transforms:
  lua_transform:
    type: lua
    inputs: [ "app_logs_src" ]
    version: "2"
    hooks:
      init: |
        function (emit)
         
          _G.event_count = 0
          _G.max_events_per_batch = 5000
          _G.sleep_time = 2  
        end
      process: |
        function (event, emit)
          
          _G.event_count = _G.event_count + 1
        
          --print("_G.event_count: " .. _G.event_count)
          --print("Processing event: " .. tostring(event.log))
        
          if _G.event_count >= _G.max_events_per_batch then
           
            os.execute("sleep " .. _G.sleep_time)
            _G.event_count = 0  
          end
          emit(event)
        end
  app_logs_parse:
    type: "remap"
    inputs: ["lua_transform"]  
    source: |
      . |= parse_regex!(.message, r'^\[(?P<bizType>[^\]]+)\] \[(?P<appName>[^\]]+)\] \[(?P<traceId>[^\]]*)\] \[(?P<ipAddress>[^\]]+)\] \[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}:\d{3})\] \[(?P<level>[^\]]+)\] \[(?P<threadId>[^\]]+)\] - (?P<message>[\s\S]*)$')            
      del(.host)
      del(.source_type)
      del(.file)
    drop_on_error: true
    drop_on_abort: true
    reroute_dropped: true
sinks:
  dropped_sink:
    inputs: [ "app_logs_parse.dropped" ]
    type: file
    path: "/alidata/vector/testScript/vector_error.log"
    encoding:
      codec: json
  temp_log_test_sink:
    inputs:
      - app_logs_parse
    type: file
    path: "/alidata/vector/testScript/createLog/output.log"
    encoding:
      codec: json
      json:
        pretty: true
  web_log_es_sink:
    type: "elasticsearch"
    inputs: ["app_logs_parse"]
    endpoints:
      - "http://172.18.83.221:9201"
    api_version: v7
    bulk:
      index: "application_log.%Y-%m-%d"
    batch:
      max_events: 5000
      timeout_secs: 5
    buffer:
      type: "disk"
      #max_size: 269484544
      max_size: 2269484544
      when_full: "block"

[xufeixianggithub]

I provided a piece of code using lua stream limiting, tested, it can control the speed of three components, source,transforms,sink it can be implemented, read only 5K logs at a time, and tested, the number of logs output and the number of logs in the source file is consistent, the cpu usage is only used a few points. If anyone has seen or tried this solution, they can point out whether it has hidden dangers.

[pront]

Hi @xufeixianggithub, thank you sharing all this context. I wonder if something like this new transform #21071 would be useful to you.