fix(ingest) write fs error to events log

fix(operator) forbid fetch local from local IPs
fix(functions-server) get fetch timeout from rotor

Author: absorbb

bulker/ingest/router.go

@@ -117,7 +117,7 @@ func NewRouter(appContext *Context, partitionSelector kafkabase.PartitionSelecto
 	})
 
 	httpClient := &http.Client{
-		Timeout: time.Duration(appContext.config.DeviceFunctionsTimeoutMs) * time.Millisecond,
+		Timeout: time.Duration(int64(float64(appContext.config.DeviceFunctionsTimeoutMs)*1.1)) * time.Millisecond,
 	}
 
 	var dataHosts []string
@@ -494,7 +494,7 @@ func (r *Router) processSyncDestination(message *IngestMessage, stream *StreamWi
 			// Use functions server (new format with execLog)
 			fsURL := getFunctionsServerURL(r.config.FunctionsServerURLTemplate, stream.Stream.WorkspaceId, classes)
 			endpointURL := fsURL + "/multi"
-			r.callFunctionsEndpoint(functionDestinations, endpointURL, messageBytes, functionsResults)
+			r.callFunctionsEndpoint(stream, functionDestinations, endpointURL, messageBytes, functionsResults)
 		} else {
 			// Use rotor (legacy format)
 			endpointURL := r.config.RotorURL + "/func/multi"
@@ -606,19 +606,23 @@ type ConnectionChainResult struct {
 
 // callFunctionsEndpoint sends a request to functions endpoint and expects new format with execLog
 // Response format: map[connectionId]{ events: [], execLog: [] }
-func (r *Router) callFunctionsEndpoint(destinations []*ShortDestinationConfig, baseURL string, messageBytes []byte, functionsResults map[string]any) {
+func (r *Router) callFunctionsEndpoint(stream *StreamWithDestinations, destinations []*ShortDestinationConfig, baseURL string, messageBytes []byte, functionsResults map[string]any) {
 	if len(destinations) == 0 {
 		return
 	}
 
 	ids := utils.ArrayMap(destinations, func(d *ShortDestinationConfig) string { return d.ConnectionId })
 	var err error
 	defer func() {
-		for _, id := range ids {
-			if err != nil {
+		if err != nil {
+			obj := map[string]any{"body": string(messageBytes), "error": "Functions server ", "status": "FS_ERROR"}
+			r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: stream.Stream.Id, Event: obj})
+			for _, id := range ids {
 				DeviceFunctions(id, "error").Inc()
 				DeviceFunctions("total", "error").Inc()
-			} else {
+			}
+		} else {
+			for _, id := range ids {
 				DeviceFunctions(id, "success").Inc()
 				DeviceFunctions("total", "success").Inc()
 			}
@@ -641,19 +645,22 @@ func (r *Router) callFunctionsEndpoint(destinations []*ShortDestinationConfig, b
 	res, err := r.httpClient.Do(req)
 	if err != nil {
 		r.Errorf("failed to send functions request for connections: %s: %v", ids, err)
+		err = fmt.Errorf("functions request error: %v", err)
 		return
 	}
 	defer res.Body.Close()
 
 	body, err := io.ReadAll(res.Body)
 	if res.StatusCode != 200 || err != nil {
 		r.Errorf("Failed to send functions request for connections: %s: status: %v body: %s", ids, res.StatusCode, string(body))
+		err = fmt.Errorf("functions response error: status %d err: %v", res.StatusCode, err)
 		return
 	}
 	var result map[string]ConnectionChainResult
 	err = jsoniter.Unmarshal(body, &result)
 	if err != nil {
 		r.Errorf("Failed to unmarshal functions response for connections: %s: %v", ids, err)
+		err = fmt.Errorf("functions response error: %v", err)
 		return
 	}
 

bulker/operator/operator.go

@@ -1226,6 +1226,10 @@ func (o *Operator) buildDeploymentFromData(data *DeploymentData) *appsv1.Deploym
 			Name:  "FUNCTIONS_CLASS",
 			Value: data.FunctionsClass,
 		},
+		{
+			Name:  "FETCH_FORBID_LOCAL",
+			Value: "true",
+		},
 	}
 
 	// Build containers list

services/rotor/src/functions-server.ts

@@ -988,7 +988,7 @@ async function main() {
       const eventContext = createEventContextFromMessage(message, connection, 0);
       const functionsFetchTimeout = req.headers["x-request-timeout-ms"]
         ? parseNumber(req.headers["x-request-timeout-ms"] as string, 2000)
-        : 2000;
+        : parseNumber(env.FETCH_TIMEOUT_MS, 2000);
       try {
         const result = await runChain(chain, event, eventContext, functionsFetchTimeout);
 
@@ -1071,7 +1071,11 @@ async function main() {
       ...customContext,
     } as EventContext & { retries?: number };
 
-    const result = await runChain(chain, event, eventContext, parseNumber(env.FETCH_TIMEOUT_MS, 2000));
+    const functionsFetchTimeout = req.headers["x-request-timeout-ms"]
+      ? parseNumber(req.headers["x-request-timeout-ms"] as string, 2000)
+      : parseNumber(env.FETCH_TIMEOUT_MS, 2000);
+
+    const result = await runChain(chain, event, eventContext, functionsFetchTimeout);
 
     const totalMs = result.execLog.reduce((sum, e) => sum + (e.ms || 0), 0);
     log.atInfo().log(`← ${connectionId} (${chain.functions.length} functions) completed in ${totalMs}ms`);

services/rotor/src/lib/functions-chain.ts

@@ -300,7 +300,8 @@ export function buildFunctionChain(
       functionsClass,
       chainCtx,
       funcCtx,
-      eventsLogger
+      eventsLogger,
+      fetchTimeoutMs
     );
     return async (event: AnyEvent, ctx: EventContext) => {
       return wrapper(event, ctx);

services/rotor/src/lib/functions-server-client.ts

@@ -130,19 +130,19 @@ export async function callFunctionsServer(
   eventContext: EventContext,
   chainCtx: FunctionChainContext,
   funcCtx: FunctionContext,
-  eventsLogger: EventsStore
+  eventsLogger: EventsStore,
+  fetchTimeoutMs?: number
 ): Promise<FunctionsServerResult> {
   const serverEnv = getServerEnv();
   const url = getFunctionsServerUrl(workspaceId, connectionId, functionsClass);
   const timeoutMs = parseInt(serverEnv.FUNCTIONS_SERVER_TIMEOUT_MS);
 
-  const controller = new AbortController();
-  const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
   try {
     const response = await fetch(url, {
       method: "POST",
       headers: {
         "Content-Type": "application/json",
+        ...(fetchTimeoutMs ? { "x-request-timeout-ms": String(fetchTimeoutMs) } : {}),
       },
       body: JSON.stringify({
         event,
@@ -156,7 +156,7 @@ export async function callFunctionsServer(
           retries: eventContext.retries ?? 0,
         },
       }),
-      signal: controller.signal,
+      signal: AbortSignal.timeout(timeoutMs),
     });
 
     if (!response.ok) {
@@ -202,8 +202,6 @@ export async function callFunctionsServer(
     }
     chainCtx.metrics?.status("functions_server", "error", "other").inc(1);
     throw new RetryError(`Functions processing failed: ${e.message}`);
-  } finally {
-    clearTimeout(timeoutId);
   }
 }
 
@@ -217,7 +215,8 @@ export function createFunctionsServerWrapper(
   functionsClass: Omit<FunctionsClass, "legacy">,
   chainCtx: FunctionChainContext,
   funcCtx: FunctionContext,
-  eventsLogger: EventsStore
+  eventsLogger: EventsStore,
+  fetchTimeoutMs?: number
 ): (event: AnyEvent, ctx: EventContext) => Promise<AnyEvent | AnyEvent[] | "drop" | undefined> {
   return async (event: AnyEvent, ctx: EventContext) => {
     try {
@@ -229,7 +228,8 @@ export function createFunctionsServerWrapper(
         ctx,
         chainCtx,
         funcCtx,
-        eventsLogger
+        eventsLogger,
+        fetchTimeoutMs
       );
 
       // Check for errors in execLog - similar to checkError in udf-wrapper-code.txtjs