in_opentelemetry: process eligible logs in HTTP workers

include/fluent-bit/flb_downstream.h

@@ -98,6 +98,18 @@ int flb_downstream_conn_event_accept(
         void *accept_callback_data,
         flb_connection_event_callback event_callback,
         int mask);
+
+/*
+ * Same as flb_downstream_conn_event_accept() for a socket that was already
+ * accepted, for example by another worker. The socket is always consumed.
+ */
+int flb_downstream_conn_event_adopt(
+        struct flb_downstream *stream,
+        flb_sockfd_t fd,
+        flb_connection_accept_callback accept_callback,
+        void *accept_callback_data,
+        flb_connection_event_callback event_callback,
+        int mask);
 int flb_downstream_conn_event_register(struct flb_connection *connection,
                                        int (*callback)(void *data),
                                        int mask);

include/fluent-bit/flb_input.h

@@ -957,6 +957,12 @@ int flb_input_ingress_queue_log_take_records(struct flb_input_instance *ins,
                                              const char *tag, size_t tag_len,
                                              void *buf, size_t buf_size,
                                              size_t allocation_size);
+int flb_input_ingress_queue_log_processed_take_records(
+                                             struct flb_input_instance *ins,
+                                             size_t records,
+                                             const char *tag, size_t tag_len,
+                                             void *buf, size_t buf_size,
+                                             size_t allocation_size);
 int flb_input_ingress_queue_metrics(struct flb_input_instance *ins,
                                     const char *tag, size_t tag_len,
                                     struct cmt *cmt, size_t payload_size);

include/fluent-bit/flb_input_log.h

@@ -45,4 +45,18 @@ int flb_input_log_append_processed(struct flb_input_instance *ins,
                                    size_t tag_len,
                                    const void *buf,
                                    size_t buf_size);
+int flb_input_log_append_processed_routed(struct flb_input_instance *ins,
+                                          size_t records,
+                                          const char *tag,
+                                          size_t tag_len,
+                                          const void *buf,
+                                          size_t buf_size);
+
+struct flb_mp_chunk_cobj;
+int flb_input_log_chunk_supported(struct flb_input_instance *ins,
+                                  int worker_thread);
+int flb_input_log_append_chunk(struct flb_input_instance *ins,
+                               const char *tag, size_t tag_len,
+                               struct flb_mp_chunk_cobj *chunk,
+                               int worker_thread);
 #endif