From 7317c8b82b8e8b779cf81265774f6253f013c234 Mon Sep 17 00:00:00 2001 From: Ming Ho Date: Mon, 17 Aug 2026 13:49:29 -0700 Subject: [PATCH] CBD-6545, templates and scripts to generate enterprise-analytics-udf image These are used to generate dockerfile and to publish image to internal and public registries. -Ming --- generate/generator/generate.go | 24 +- .../enterprise-analytics-udf/README.md | 1 + .../scripts/udf_listener.py | 278 ++++++++++++++++++ .../Dockerfile.template | 33 +++ 4 files changed, 329 insertions(+), 7 deletions(-) create mode 100644 generate/resources/enterprise-analytics-udf/README.md create mode 100644 generate/resources/enterprise-analytics-udf/scripts/udf_listener.py create mode 100644 generate/templates/enterprise-analytics-udf/Dockerfile.template diff --git a/generate/generator/generate.go b/generate/generator/generate.go index d4d08ae4..b1d5ade3 100644 --- a/generate/generator/generate.go +++ b/generate/generator/generate.go @@ -33,12 +33,13 @@ const ( type Product string const ( - ProductServer = Product("couchbase-server") - ProductSyncGw = Product("sync-gateway") - ProductSandbox = Product("server-sandbox") - ProductColumnar = Product("couchbase-columnar") - ProductEdgeServer = Product("couchbase-edge-server") - ProductEnterpriseAnalytics = Product("enterprise-analytics") + ProductServer = Product("couchbase-server") + ProductSyncGw = Product("sync-gateway") + ProductSandbox = Product("server-sandbox") + ProductColumnar = Product("couchbase-columnar") + ProductEdgeServer = Product("couchbase-edge-server") + ProductEnterpriseAnalytics = Product("enterprise-analytics") + ProductEnterpriseAnalyticsUdf = Product("enterprise-analytics-udf") ) // These are Docker's idea of architecture names, eg. amd64, arm64. @@ -99,6 +100,7 @@ func init() { ProductColumnar, ProductEdgeServer, ProductEnterpriseAnalytics, + ProductEnterpriseAnalyticsUdf, } // TODO: Read the version_customizations.json file into map @@ -245,7 +247,7 @@ func generateOneDockerfile( // 7.1.0 and higher also support arm64 variant.Arches = append(variant.Arches, Archarm64) } - } else if product == ProductColumnar || product == ProductEnterpriseAnalytics { + } else if product == ProductColumnar || product == ProductEnterpriseAnalytics || product == ProductEnterpriseAnalyticsUdf { variant.Arches = append(variant.Arches, Archarm64) } @@ -358,6 +360,12 @@ func generateDockerfile(variant DockerfileVariant) error { "DOCKER_BASE_IMAGE": variant.dockerBaseImage(), "CB_MULTIARCH": len(variant.Arches) > 1, } + } else if variant.Product == ProductEnterpriseAnalyticsUdf { + // No Couchbase package: the UDF executor image is built entirely + // from OS packages on a fixed base image, independent of version. + params = map[string]any{ + "DOCKER_BASE_IMAGE": variant.dockerBaseImage(), + } } else if variant.Product == ProductEdgeServer { // template parameters params = map[string]any{ @@ -597,6 +605,8 @@ func (variant DockerfileVariant) dockerBaseImage() string { return fmt.Sprintf("ubuntu:%s", variant.ubuntuVersion()) case ProductEnterpriseAnalytics: return fmt.Sprintf("ubuntu:%s", variant.ubuntuVersion()) + case ProductEnterpriseAnalyticsUdf: + return "debian:12-slim" default: log.Printf("Failed %v", variant.Product) panic("Unexpected product") diff --git a/generate/resources/enterprise-analytics-udf/README.md b/generate/resources/enterprise-analytics-udf/README.md new file mode 100644 index 00000000..a14c13fc --- /dev/null +++ b/generate/resources/enterprise-analytics-udf/README.md @@ -0,0 +1 @@ +# Enterprise Analytics UDF Executor diff --git a/generate/resources/enterprise-analytics-udf/scripts/udf_listener.py b/generate/resources/enterprise-analytics-udf/scripts/udf_listener.py new file mode 100644 index 00000000..49069de5 --- /dev/null +++ b/generate/resources/enterprise-analytics-udf/scripts/udf_listener.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import signal +import socket +import sys +import traceback +from enum import IntEnum +from importlib import import_module +from io import BytesIO +from os import chdir +from os import getcwd +from os import getpid +from pathlib import Path +from struct import pack, unpack + +import msgpack + + +PROTO_VERSION = 1 +HEADER_SZ = 8 + 8 + 1 +REAL_HEADER_SZ = 4 + 8 + 8 + 1 +FRAMESZ = 32768 + + +class MessageType(IntEnum): + HELO = 0 + QUIT = 1 + INIT = 2 + INIT_RSP = 3 + CALL = 4 + CALL_RSP = 5 + ERROR = 6 + + +class MessageFlags(IntEnum): + NORMAL = 0 + INITIAL_REQ = 1 + INITIAL_ACK = 2 + ERROR = 3 + + +class Wrapper(object): + wrapped_module = None + wrapped_class = None + wrapped_fn = None + sz = None + mid = None + rmid = None + flag = None + resp = None + unpacked_msg = None + msg_type = None + packer = msgpack.Packer(autoreset=False, use_bin_type=False) + unpacker = msgpack.Unpacker(raw=False) + response_buf = BytesIO() + stdin_buf = BytesIO() + wrapped_fns = {} + alive = True + readbuf = bytearray(FRAMESZ) + readview = memoryview(readbuf) + + def init(self, module_name, class_name, fn_name): + self.wrapped_module = import_module(module_name) + wrapped_fn = None + if not self.check_module_path(self.wrapped_module): + self.wrapped_module = None + raise ImportError("Module was not found in library") + if class_name is not None: + self.wrapped_class = getattr( + import_module(module_name), class_name)() + if self.wrapped_class is not None: + wrapped_fn = getattr(self.wrapped_class, fn_name) + else: + wrapped_fn = getattr(import_module(module_name), fn_name) + if wrapped_fn is None: + raise ImportError( + "Could not find class or function in specified module") + self.wrapped_fns[self.mid] = wrapped_fn + + def next_tuple(self, *args, key=None): + return self.wrapped_fns[key](*args) + + def check_module_path(self, module): + cwd = Path('.').resolve() + module_path = Path(module.__file__).resolve() + return cwd in module_path.parents + + def read_header(self, readbuf): + self.sz, self.mid, self.rmid, self.flag = unpack( + "!iqqb", readbuf[0:REAL_HEADER_SZ]) + return True + + def write_header(self, response_buf, dlen): + total_len = dlen + HEADER_SZ + header = pack("!iqqb", total_len, int(-1), int(self.rmid), self.flag) + self.response_buf.write(header) + return total_len + 4 + + def get_ver_hlen(self, hlen): + return hlen + (PROTO_VERSION << 4) + + def get_hlen(self): + return self.ver_hlen - (PROTO_VERSION << 4) + + def init_remote_ipc(self): + self.response_buf.seek(0) + self.flag = MessageFlags.INITIAL_REQ + dlen = len(self.unpacked_msg[1]) + resp_len = self.write_header(self.response_buf, dlen) + self.response_buf.write(self.unpacked_msg[1]) + self.resp = self.response_buf.getbuffer()[0:resp_len] + self.send_msg() + self.packer.reset() + + def cd(self, basedir): + chdir(basedir + "/site-packages") + sys.path.insert(0, getcwd()) + + def helo(self): + self.cd(self.unpacked_msg[1][1]) + self.flag = MessageFlags.NORMAL + self.response_buf.seek(0) + self.packer.pack(int(MessageType.HELO)) + self.packer.pack(int(getpid())) + dlen = len(self.packer.bytes()) + resp_len = self.write_header(self.response_buf, dlen) + self.response_buf.write(self.packer.bytes()) + self.resp = self.response_buf.getbuffer()[0:resp_len] + self.send_msg() + self.packer.reset() + return True + + def handle_init(self): + self.flag = MessageFlags.NORMAL + self.response_buf.seek(0) + args = self.unpacked_msg[1] + module = args[0] + if len(args) == 3: + clazz = args[1] + fn = args[2] + else: + clazz = None + fn = args[1] + self.init(module, clazz, fn) + self.packer.pack(int(MessageType.INIT_RSP)) + dlen = 1 + resp_len = self.write_header(self.response_buf, dlen) + self.response_buf.write(self.packer.bytes()) + self.resp = self.response_buf.getbuffer()[0:resp_len] + self.send_msg() + self.packer.reset() + return True + + def quit(self): + self.alive = False + return True + + def handle_call(self): + self.flag = MessageFlags.NORMAL + result = ([], []) + if len(self.unpacked_msg) > 1: + args = self.unpacked_msg[1] + if args is not None: + for arg in args: + try: + result[0].append(self.next_tuple(*arg, key=self.mid)) + except BaseException: + result[1].append(traceback.format_exc()) + self.packer.reset() + self.response_buf.seek(0) + body = msgpack.packb(result) + dlen = len(body) + 1 + resp_len = self.write_header(self.response_buf, dlen) + self.packer.pack(int(MessageType.CALL_RSP)) + self.response_buf.write(self.packer.bytes()) + self.response_buf.write(body) + self.resp = self.response_buf.getbuffer()[0:resp_len] + self.send_msg() + self.packer.reset() + return True + + def handle_error(self, error): + self.flag = MessageFlags.NORMAL + self.packer.reset() + self.response_buf.seek(0) + body = msgpack.packb(str(error)) + dlen = len(body) + 1 + resp_len = self.write_header(self.response_buf, dlen) + self.packer.pack(int(MessageType.ERROR)) + self.response_buf.write(self.packer.bytes()) + self.response_buf.write(body) + self.resp = self.response_buf.getbuffer()[0:resp_len] + self.send_msg() + self.packer.reset() + self.alive = False + return True + + type_handler = { + MessageType.HELO: helo, + MessageType.QUIT: quit, + MessageType.INIT: handle_init, + MessageType.CALL: handle_call, + } + + def connect_sock(self): + self.sock = socket.fromfd(1, socket.AF_UNIX, socket.SOCK_STREAM) + + def disconnect_sock(self, *args): + self.sock.shutdown(socket.SHUT_RDWR) + self.sock.close() + + def recv_msg(self): + while self.alive: + pos = self.sock.recv_into(self.readbuf) + if pos <= 0: + self.alive = False + return + try: + while pos < REAL_HEADER_SZ: + read = self.sock.recv_into(self.readview[pos:]) + if read <= 0: + self.alive = False + return + pos += read + self.read_header(self.readview) + while pos < self.sz and len(self.readbuf) - pos > 0: + read = self.sock.recv_into(self.readview[pos:]) + if read <= 0: + self.alive = False + return + pos += read + while pos < self.sz: + vszchunk = self.sock.recv(4096) + if len(vszchunk) == 0: + self.alive = False + return + self.readview.release() + self.readbuf.extend(vszchunk) + self.readview = memoryview(self.readbuf) + pos += len(vszchunk) + self.unpacker.feed(self.readview[REAL_HEADER_SZ:self.sz]) + self.unpacked_msg = list(self.unpacker) + self.msg_type = MessageType(self.unpacked_msg[0]) + self.type_handler[self.msg_type](self) + except BaseException: + self.handle_error(''.join(traceback.format_exc())) + + def send_msg(self): + self.sock.sendall(self.resp) + self.resp = None + return + + def recv_loop(self): + while self.alive: + self.recv_msg() + self.disconnect_sock() + + +wrap = Wrapper() +wrap.connect_sock() +signal.signal(signal.SIGTERM, wrap.disconnect_sock) +wrap.recv_loop() diff --git a/generate/templates/enterprise-analytics-udf/Dockerfile.template b/generate/templates/enterprise-analytics-udf/Dockerfile.template new file mode 100644 index 00000000..7b49e1cb --- /dev/null +++ b/generate/templates/enterprise-analytics-udf/Dockerfile.template @@ -0,0 +1,33 @@ +FROM {{ .DOCKER_BASE_IMAGE }} + +LABEL maintainer="docker@couchbase.com" + +ENV TZ=Etc/UTC +ENV DEBIAN_FRONTEND=noninteractive +COPY ../../resources/enterprise-analytics-udf/scripts/udf_listener.py . +RUN apt update &&\ + apt -y upgrade && \ + apt -y install \ + apt-transport-https \ + ca-certificates \ + curl \ + gnupg \ + python3 \ + python3-msgpack \ + s6 \ + vim \ + wget \ + procps \ + sudo +RUN set -eux; \ + if ! getent group udf-executor >/dev/null 2>&1; then \ + groupadd --system udf-executor; \ + fi; \ + if ! id -u udf-executor >/dev/null 2>&1; then \ + useradd --system --home-dir /tmp --gid udf-executor udf-executor; \ + fi; \ + chown udf-executor:udf-executor /udf_listener.py; \ + mkdir -p /var/run/udf-sockets; \ + chown udf-executor:udf-executor /var/run/udf-sockets +USER udf-executor:udf-executor +ENTRYPOINT ["s6-ipcserver", "-v", "-a", "0700", "/var/run/udf-sockets/pyudf.socket", "python3", "/udf_listener.py"]