CBD-6545, templates and scripts to generate enterprise-analytics-udf image

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")

generate/resources/enterprise-analytics-udf/README.md

@@ -0,0 +1 @@
+# Enterprise Analytics UDF Executor

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()

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"]