Compare commits

..
1 Commits
Author SHA1 Message Date
opencodeandwoggioni 5b52677c28 Generalize OTEL API and add memcache tracing support
CI / build (push) Successful in 3m28s
- Rename RedisSpan -> SpanHandle for generic span handling
- Generalize TelemetryController methods: startSpan/endSpan with dbSystem param
- Rename RedisOtelSpan -> OtelSpanHandle in rbcs-server-otel
- Update Redis cache handler to use new generic API
- Add OpenTelemetry tracing for memcache GET and SET commands
- Add channel property to MemcacheRequestController for server address attribution
- Add uses TelemetryController directive in memcache module-info

Memcache spans follow the same pattern as Redis:
db.system=memcache, db.operation=GET|SET, server.address, server.port
2026-05-23 23:46:41 +08:00
20 changed files with 94 additions and 1232 deletions
-6
View File
@@ -28,7 +28,6 @@ jobs:
with: with:
builder: "multiplatform-builder" builder: "multiplatform-builder"
context: "docker/build/docker" context: "docker/build/docker"
build-args: VERSION=${{ steps.retrieve-version.outputs.VERSION }},REVISION=${{ github.sha }}
platforms: linux/amd64,linux/arm64 platforms: linux/amd64,linux/arm64
push: true push: true
pull: true pull: true
@@ -42,7 +41,6 @@ jobs:
with: with:
builder: "multiplatform-builder" builder: "multiplatform-builder"
context: "docker/build/docker" context: "docker/build/docker"
build-args: VERSION=${{ steps.retrieve-version.outputs.VERSION }},REVISION=${{ github.sha }}
platforms: linux/amd64,linux/arm64 platforms: linux/amd64,linux/arm64
push: true push: true
pull: true pull: true
@@ -56,7 +54,6 @@ jobs:
with: with:
builder: "multiplatform-builder" builder: "multiplatform-builder"
context: "docker/build/docker" context: "docker/build/docker"
build-args: VERSION=${{ steps.retrieve-version.outputs.VERSION }},REVISION=${{ github.sha }}
platforms: linux/amd64,linux/arm64 platforms: linux/amd64,linux/arm64
push: true push: true
pull: true pull: true
@@ -70,7 +67,6 @@ jobs:
with: with:
builder: "multiplatform-builder" builder: "multiplatform-builder"
context: "docker/build/docker" context: "docker/build/docker"
build-args: VERSION=${{ steps.retrieve-version.outputs.VERSION }},REVISION=${{ github.sha }}
platforms: linux/amd64,linux/arm64 platforms: linux/amd64,linux/arm64
push: true push: true
pull: true pull: true
@@ -84,7 +80,6 @@ jobs:
with: with:
builder: "multiplatform-builder" builder: "multiplatform-builder"
context: "docker/build/docker" context: "docker/build/docker"
build-args: VERSION=${{ steps.retrieve-version.outputs.VERSION }},REVISION=${{ github.sha }}
platforms: linux/amd64 platforms: linux/amd64
push: true push: true
pull: true pull: true
@@ -98,7 +93,6 @@ jobs:
with: with:
builder: "multiplatform-builder" builder: "multiplatform-builder"
context: "docker/build/docker" context: "docker/build/docker"
build-args: VERSION=${{ steps.retrieve-version.outputs.VERSION }},REVISION=${{ github.sha }}
platforms: linux/amd64 platforms: linux/amd64
push: true push: true
pull: true pull: true
+2
View File
@@ -4,6 +4,8 @@
# Ignore Gradle build output directory # Ignore Gradle build output directory
build build
rbcs-cli/native-image/*.json
# Ignore JDTLS files # Ignore JDTLS files
.classpath .classpath
.project .project
-23
View File
@@ -1,25 +1,14 @@
ARG VERSION, REVISION
FROM eclipse-temurin:25-jre-alpine AS base-release FROM eclipse-temurin:25-jre-alpine AS base-release
LABEL org.opencontainers.image.authors="Walter Oggioni <walter.oggioni@agentmail.to>"
LABEL org.opencontainers.image.version="${VERSION}"
LABEL org.opencontainers.image.revision="${REVISION}"
LABEL org.opencontainers.image.source=https://gitea.woggioni.net/woggioni/rbcs
RUN adduser -D rbcs RUN adduser -D rbcs
USER rbcs USER rbcs
ENV RBCS_CONFIGURATION_DIR="/etc/rbcs"
WORKDIR /var/lib/rbcs WORKDIR /var/lib/rbcs
FROM base-release AS release-vanilla FROM base-release AS release-vanilla
LABEL org.opencontainers.image.title=rbcs
LABEL org.opencontainers.image.description=RBCS vanilla image
ADD rbcs-cli-envelope-*.jar rbcs.jar ADD rbcs-cli-envelope-*.jar rbcs.jar
ADD logback.xml /etc/rbcs/logback.xml ADD logback.xml /etc/rbcs/logback.xml
ENTRYPOINT ["java", "-jar", "/var/lib/rbcs/rbcs.jar"] ENTRYPOINT ["java", "-jar", "/var/lib/rbcs/rbcs.jar"]
FROM base-release AS release-memcache FROM base-release AS release-memcache
LABEL org.opencontainers.image.title=rbcs-memcache
LABEL org.opencontainers.image.description=RBCS image with memcache plugin
ADD --chown=rbcs:rbcs rbcs-cli-envelope-*.jar rbcs.jar ADD --chown=rbcs:rbcs rbcs-cli-envelope-*.jar rbcs.jar
RUN mkdir plugins RUN mkdir plugins
WORKDIR /var/lib/rbcs/plugins WORKDIR /var/lib/rbcs/plugins
@@ -29,8 +18,6 @@ ADD logback.xml /etc/rbcs/logback.xml
ENTRYPOINT ["java", "-jar", "/var/lib/rbcs/rbcs.jar"] ENTRYPOINT ["java", "-jar", "/var/lib/rbcs/rbcs.jar"]
FROM base-release AS release-redis FROM base-release AS release-redis
LABEL org.opencontainers.image.title=rbcs-redis
LABEL org.opencontainers.image.description=RBCS image with redis plugin
ADD --chown=rbcs:rbcs rbcs-cli-envelope-*.jar rbcs.jar ADD --chown=rbcs:rbcs rbcs-cli-envelope-*.jar rbcs.jar
RUN mkdir plugins RUN mkdir plugins
WORKDIR /var/lib/rbcs/plugins WORKDIR /var/lib/rbcs/plugins
@@ -40,8 +27,6 @@ ADD logback.xml /etc/rbcs/logback.xml
ENTRYPOINT ["java", "-jar", "/var/lib/rbcs/rbcs.jar"] ENTRYPOINT ["java", "-jar", "/var/lib/rbcs/rbcs.jar"]
FROM base-release AS release-full FROM base-release AS release-full
LABEL org.opencontainers.image.title=rbcs-full
LABEL org.opencontainers.image.description=RBCS image with all plugins
ADD --chown=rbcs:rbcs rbcs-cli-envelope-*.jar rbcs.jar ADD --chown=rbcs:rbcs rbcs-cli-envelope-*.jar rbcs.jar
RUN mkdir plugins RUN mkdir plugins
WORKDIR /var/lib/rbcs/plugins WORKDIR /var/lib/rbcs/plugins
@@ -59,22 +44,15 @@ RUN adduser -D -u 1000 rbcs -h /var/lib/rbcs
RUN chown rbcs:rbcs /var/tmp/rbcs RUN chown rbcs:rbcs /var/tmp/rbcs
FROM scratch AS release-native FROM scratch AS release-native
LABEL org.opencontainers.image.title=rbcs-native
LABEL org.opencontainers.image.description=RBCS image with a native executable with GraalVM
COPY --from=base-native /etc/passwd /etc/passwd COPY --from=base-native /etc/passwd /etc/passwd
COPY --from=base-native /etc/rbcs /etc/rbcs COPY --from=base-native /etc/rbcs /etc/rbcs
COPY --from=base-native /var/lib/rbcs /var/lib/rbcs COPY --from=base-native /var/lib/rbcs /var/lib/rbcs
COPY --from=base-native /var/tmp/rbcs /var/tmp/rbcs
ADD rbcs-cli.upx /usr/bin/rbcs-cli ADD rbcs-cli.upx /usr/bin/rbcs-cli
ADD logback.xml /etc/rbcs/logback.xml
USER rbcs USER rbcs
WORKDIR /var/lib/rbcs WORKDIR /var/lib/rbcs
ENV RBCS_CONFIGURATION_DIR="/etc/rbcs"
ENTRYPOINT ["/usr/bin/rbcs-cli", "-XX:MaximumHeapSizePercent=70", "-Dio.netty.tmpdir=/var/tmp/rbcs", "-Dlogback.configurationFile=/etc/rbcs/logback.xml"] ENTRYPOINT ["/usr/bin/rbcs-cli", "-XX:MaximumHeapSizePercent=70", "-Dio.netty.tmpdir=/var/tmp/rbcs", "-Dlogback.configurationFile=/etc/rbcs/logback.xml"]
FROM debian:12-slim AS release-jlink FROM debian:12-slim AS release-jlink
LABEL org.opencontainers.image.title=rbcs-jlink
LABEL org.opencontainers.image.description=RBCS image with a jlink distribution
RUN mkdir -p /usr/share/java/rbcs RUN mkdir -p /usr/share/java/rbcs
RUN --mount=type=bind,source=.,target=/build/distributions tar -xf /build/distributions/rbcs-cli*.tar -C /usr/share/java/rbcs RUN --mount=type=bind,source=.,target=/build/distributions tar -xf /build/distributions/rbcs-cli*.tar -C /usr/share/java/rbcs
RUN chmod 755 /usr/share/java/rbcs/bin/* RUN chmod 755 /usr/share/java/rbcs/bin/*
@@ -83,5 +61,4 @@ RUN adduser -u 1000 rbcs
USER rbcs USER rbcs
WORKDIR /var/lib/rbcs WORKDIR /var/lib/rbcs
ADD logback.xml /etc/rbcs/logback.xml ADD logback.xml /etc/rbcs/logback.xml
ENV RBCS_CONFIGURATION_DIR="/etc/rbcs"
ENTRYPOINT ["/usr/local/bin/rbcs-cli"] ENTRYPOINT ["/usr/local/bin/rbcs-cli"]
+2 -2
View File
@@ -2,9 +2,9 @@ org.gradle.configuration-cache=false
org.gradle.parallel=true org.gradle.parallel=true
org.gradle.caching=true org.gradle.caching=true
rbcs.version = 0.5.1 rbcs.version = 0.5.0
lys.version = 2026.06.08 lys.version = 2026.05.16
gitea.maven.url = https://gitea.woggioni.net/api/packages/woggioni/maven gitea.maven.url = https://gitea.woggioni.net/api/packages/woggioni/maven
docker.registry.url=gitea.woggioni.net docker.registry.url=gitea.woggioni.net
@@ -7,7 +7,4 @@ public interface SpanHandle {
void setAttribute(@NotNull String key, @NotNull String value); void setAttribute(@NotNull String key, @NotNull String value);
void setAttribute(@NotNull String key, long value); void setAttribute(@NotNull String key, long value);
void setAttribute(@NotNull String key, boolean value);
} }
@@ -4,13 +4,11 @@ import io.netty.channel.ChannelHandler;
import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable; import org.jetbrains.annotations.Nullable;
import java.util.Map;
public interface TelemetryController { public interface TelemetryController {
void initialize(); void initialize();
@NotNull ChannelHandler createHandler(); @NotNull ChannelHandler createHandler();
@Nullable SpanHandle startSpan(@NotNull String command); @Nullable SpanHandle startSpan(@NotNull String command, @NotNull String key, @NotNull String dbSystem);
void endSpan(@Nullable SpanHandle span); void endSpan(@Nullable SpanHandle span);
+5 -5
View File
@@ -90,7 +90,7 @@ Provider<EnvelopeJarTask> envelopeJarTaskProvider = tasks.named(EnvelopePlugin.E
tasks.named(NativeImagePlugin.CONFIGURE_NATIVE_IMAGE_TASK_NAME, NativeImageConfigurationTask) { tasks.named(NativeImagePlugin.CONFIGURE_NATIVE_IMAGE_TASK_NAME, NativeImageConfigurationTask) {
toolchain { toolchain {
languageVersion = JavaLanguageVersion.of(25) languageVersion = JavaLanguageVersion.of(25)
vendor = JvmVendorSpec.ORACLE vendor = JvmVendorSpec.GRAAL_VM
} }
mainClass = "net.woggioni.rbcs.cli.graal.GraalNativeImageConfiguration" mainClass = "net.woggioni.rbcs.cli.graal.GraalNativeImageConfiguration"
classpath = project.files( classpath = project.files(
@@ -108,10 +108,10 @@ tasks.named(NativeImagePlugin.CONFIGURE_NATIVE_IMAGE_TASK_NAME, NativeImageConfi
nativeImage { nativeImage {
toolchain { toolchain {
languageVersion = JavaLanguageVersion.of(25) languageVersion = JavaLanguageVersion.of(25)
vendor = JvmVendorSpec.ORACLE vendor = JvmVendorSpec.GRAAL_VM
} }
mainClass = mainClassName mainClass = mainClassName
//mainModule = mainModuleName // mainModule = mainModuleName
useMusl = true useMusl = true
buildStaticImage = true buildStaticImage = true
linkAtBuildTime = false linkAtBuildTime = false
@@ -119,7 +119,6 @@ nativeImage {
compressExecutable = true compressExecutable = true
compressionLevel = 6 compressionLevel = 6
useLZMA = false useLZMA = false
//verbose = true
} }
Provider<UpxTask> upxTaskProvider = tasks.named(NativeImagePlugin.UPX_TASK_NAME, UpxTask) { Provider<UpxTask> upxTaskProvider = tasks.named(NativeImagePlugin.UPX_TASK_NAME, UpxTask) {
@@ -128,7 +127,7 @@ Provider<UpxTask> upxTaskProvider = tasks.named(NativeImagePlugin.UPX_TASK_NAME,
Provider<JlinkTask> jlinkTaskProvider = tasks.named(JlinkPlugin.JLINK_TASK_NAME, JlinkTask) { Provider<JlinkTask> jlinkTaskProvider = tasks.named(JlinkPlugin.JLINK_TASK_NAME, JlinkTask) {
toolchain { toolchain {
languageVersion = JavaLanguageVersion.of(25) languageVersion = JavaLanguageVersion.of(25)
vendor = JvmVendorSpec.ORACLE vendor = JvmVendorSpec.GRAAL_VM
} }
mainClass = mainClassName mainClass = mainClassName
@@ -153,6 +152,7 @@ Provider<JlinkTask> jlinkTaskProvider = tasks.named(JlinkPlugin.JLINK_TASK_NAME,
} }
Provider<Tar> jlinkDistTarTaskProvider = tasks.named(JlinkPlugin.JLINK_DIST_TAR_TASK_NAME, Tar) { Provider<Tar> jlinkDistTarTaskProvider = tasks.named(JlinkPlugin.JLINK_DIST_TAR_TASK_NAME, Tar) {
exclude 'lib/libjvmcicompiler.so'
} }
tasks.named(JavaPlugin.PROCESS_RESOURCES_TASK_NAME, ProcessResources) { tasks.named(JavaPlugin.PROCESS_RESOURCES_TASK_NAME, ProcessResources) {
File diff suppressed because one or more lines are too long
+1 -10
View File
@@ -1,11 +1,2 @@
Args=-O3 \ Args=-O3 -march=x86-64-v2 --gc=serial --initialize-at-run-time=io.netty --enable-url-protocols=jpms -H:+UnlockExperimentalVMOptions -H:+SharedArenaSupport --initialize-at-build-time=net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory,net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory$JpmsHandler
-march=x86-64-v3 \
--gc=serial \
--enable-url-protocols=jpms \
--pgo=conf/default.iprof \
--initialize-at-run-time=io.netty \
--initialize-at-build-time=net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory,net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory$JpmsHandler \
--trace-object-instantiation=ch.qos.logback.classic.Logger \
-H:+UnlockExperimentalVMOptions \
-H:+SharedArenaSupport
#-H:TraceClassInitialization=io.netty.handler.ssl.BouncyCastleAlpnSslUtils #-H:TraceClassInitialization=io.netty.handler.ssl.BouncyCastleAlpnSslUtils
File diff suppressed because it is too large Load Diff
@@ -10,7 +10,6 @@ import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicInteger
import javax.net.ssl.TrustManagerFactory import javax.net.ssl.TrustManagerFactory
import javax.net.ssl.X509ExtendedTrustManager
import javax.net.ssl.X509TrustManager import javax.net.ssl.X509TrustManager
import kotlin.random.Random import kotlin.random.Random
import io.netty.util.concurrent.Future as NettyFuture import io.netty.util.concurrent.Future as NettyFuture
@@ -75,25 +74,13 @@ class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoC
) )
profile.tlsTruststore?.let { trustStore -> profile.tlsTruststore?.let { trustStore ->
if (!trustStore.verifyServerCertificate) { if (!trustStore.verifyServerCertificate) {
trustManager(object : X509ExtendedTrustManager() { trustManager(object : X509TrustManager {
override fun checkClientTrusted(certChain: Array<out X509Certificate>, p1: String?) { override fun checkClientTrusted(certChain: Array<out X509Certificate>, p1: String?) {
} }
override fun checkClientTrusted(certChain: Array<out X509Certificate>, p1: String?, socket: java.net.Socket) {
}
override fun checkClientTrusted(certChain: Array<out X509Certificate>, p1: String?, engine: javax.net.ssl.SSLEngine) {
}
override fun checkServerTrusted(certChain: Array<out X509Certificate>, p1: String?) { override fun checkServerTrusted(certChain: Array<out X509Certificate>, p1: String?) {
} }
override fun checkServerTrusted(certChain: Array<out X509Certificate>, p1: String?, socket: java.net.Socket) {
}
override fun checkServerTrusted(certChain: Array<out X509Certificate>, p1: String?, engine: javax.net.ssl.SSLEngine) {
}
override fun getAcceptedIssuers() = null override fun getAcceptedIssuers() = null
}) })
} else { } else {
@@ -19,7 +19,6 @@ import java.security.cert.X509Certificate
import java.util.EnumSet import java.util.EnumSet
import java.util.ServiceLoader import java.util.ServiceLoader
import javax.net.ssl.TrustManagerFactory import javax.net.ssl.TrustManagerFactory
import javax.net.ssl.X509ExtendedTrustManager
import javax.net.ssl.X509TrustManager import javax.net.ssl.X509TrustManager
import net.woggioni.jwo.JWO import net.woggioni.jwo.JWO
import net.woggioni.jwo.Tuple2 import net.woggioni.jwo.Tuple2
@@ -125,7 +124,7 @@ object RBCS {
return keystore return keystore
} }
fun getTrustManager(trustStore: KeyStore?, certificateRevocationEnabled: Boolean): X509ExtendedTrustManager { fun getTrustManager(trustStore: KeyStore?, certificateRevocationEnabled: Boolean): X509TrustManager {
return if (trustStore != null) { return if (trustStore != null) {
val certificateFactory = CertificateFactory.getInstance("X.509") val certificateFactory = CertificateFactory.getInstance("X.509")
val validator = CertPathValidator.getInstance("PKIX").apply { val validator = CertPathValidator.getInstance("PKIX").apply {
@@ -137,7 +136,7 @@ object RBCS {
val params = PKIXParameters(trustStore).apply { val params = PKIXParameters(trustStore).apply {
isRevocationEnabled = certificateRevocationEnabled isRevocationEnabled = certificateRevocationEnabled
} }
object : X509ExtendedTrustManager() { object : X509TrustManager {
override fun checkClientTrusted(chain: Array<out X509Certificate>, authType: String) { override fun checkClientTrusted(chain: Array<out X509Certificate>, authType: String) {
val clientCertificateChain = certificateFactory.generateCertPath(chain.toList()) val clientCertificateChain = certificateFactory.generateCertPath(chain.toList())
try { try {
@@ -147,26 +146,10 @@ object RBCS {
} }
} }
override fun checkClientTrusted(chain: Array<out X509Certificate>, authType: String, socket: java.net.Socket) {
checkClientTrusted(chain, authType)
}
override fun checkClientTrusted(chain: Array<out X509Certificate>, authType: String, engine: javax.net.ssl.SSLEngine) {
checkClientTrusted(chain, authType)
}
override fun checkServerTrusted(chain: Array<out X509Certificate>, authType: String) { override fun checkServerTrusted(chain: Array<out X509Certificate>, authType: String) {
throw NotImplementedError() throw NotImplementedError()
} }
override fun checkServerTrusted(chain: Array<out X509Certificate>, authType: String, socket: java.net.Socket) {
checkServerTrusted(chain, authType)
}
override fun checkServerTrusted(chain: Array<out X509Certificate>, authType: String, engine: javax.net.ssl.SSLEngine) {
checkServerTrusted(chain, authType)
}
private val acceptedIssuers = trustStore.aliases().asSequence() private val acceptedIssuers = trustStore.aliases().asSequence()
.filter(trustStore::isCertificateEntry) .filter(trustStore::isCertificateEntry)
.map(trustStore::getCertificate) .map(trustStore::getCertificate)
@@ -178,8 +161,8 @@ object RBCS {
} }
} else { } else {
val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()) val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
trustManagerFactory.trustManagers.asSequence().filter { it is X509ExtendedTrustManager } trustManagerFactory.trustManagers.asSequence().filter { it is X509TrustManager }
.single() as X509ExtendedTrustManager .single() as X509TrustManager
} }
} }
@@ -262,17 +262,7 @@ class MemcacheCacheHandler(
val key = ctx.alloc().buffer().also { val key = ctx.alloc().buffer().also {
it.writeBytes(processCacheKey(msg.key, keyPrefix, digestAlgorithm)) it.writeBytes(processCacheKey(msg.key, keyPrefix, digestAlgorithm))
} }
val memcacheSpan = telemetryController?.startSpan("GET")?.apply { val memcacheSpan = telemetryController?.startSpan("GET", msg.key, "memcache")
setAttribute("db.system", "memcache")
setAttribute("db.operation.name", "GET")
val remoteAddr = ctx.channel().remoteAddress()
if (remoteAddr is InetSocketAddress) {
remoteAddr.hostString?.let {
setAttribute("server.address", it)
}
setAttribute("server.port", remoteAddr.port.toLong())
}
}
val responseHandler = object : MemcacheResponseHandler { val responseHandler = object : MemcacheResponseHandler {
override fun responseReceived(response: BinaryMemcacheResponse) { override fun responseReceived(response: BinaryMemcacheResponse) {
val status = response.status() val status = response.status()
@@ -328,6 +318,11 @@ class MemcacheCacheHandler(
} }
} }
client.sendRequest(key.retainedDuplicate(), responseHandler).thenAccept { requestHandle -> client.sendRequest(key.retainedDuplicate(), responseHandler).thenAccept { requestHandle ->
val remoteAddr = requestHandle.channel.remoteAddress()
if (remoteAddr is InetSocketAddress) {
remoteAddr.hostString?.let { memcacheSpan?.setAttribute("server.address", it) }
memcacheSpan?.setAttribute("server.port", remoteAddr.port.toLong())
}
log.trace(ctx) { log.trace(ctx) {
"Sending GET request for key ${msg.key} to memcache" "Sending GET request for key ${msg.key} to memcache"
} }
@@ -406,18 +401,7 @@ class MemcacheCacheHandler(
"Received last chunk of ${msg.content().readableBytes()} bytes for memcache" "Received last chunk of ${msg.content().readableBytes()} bytes for memcache"
} }
putRequest.write(msg.content()) putRequest.write(msg.content())
val memcacheSpan = telemetryController?.startSpan("SET", val memcacheSpan = telemetryController?.startSpan("SET", putRequest.entryKey, "memcache")
)?.apply {
setAttribute("db.system", "memcache")
setAttribute("db.operation.name", "SET")
val remoteAddr = ctx.channel().remoteAddress()
if (remoteAddr is InetSocketAddress) {
remoteAddr.hostString?.let {
setAttribute("server.address", it)
}
setAttribute("server.port", remoteAddr.port.toLong())
}
}
putRequest.memcacheSpanRef.set(memcacheSpan) putRequest.memcacheSpanRef.set(memcacheSpan)
val key = putRequest.digest.retainedDuplicate() val key = putRequest.digest.retainedDuplicate()
val (payloadSize, payloadSource) = putRequest.commit() val (payloadSize, payloadSource) = putRequest.commit()
@@ -2,7 +2,6 @@ package net.woggioni.rbcs.server.otel
import io.netty.channel.ChannelHandler import io.netty.channel.ChannelHandler
import io.opentelemetry.api.GlobalOpenTelemetry import io.opentelemetry.api.GlobalOpenTelemetry
import io.opentelemetry.api.common.AttributeKey
import io.opentelemetry.api.trace.SpanKind import io.opentelemetry.api.trace.SpanKind
import io.opentelemetry.api.trace.StatusCode import io.opentelemetry.api.trace.StatusCode
import io.opentelemetry.instrumentation.logback.appender.v1_0.OpenTelemetryAppender import io.opentelemetry.instrumentation.logback.appender.v1_0.OpenTelemetryAppender
@@ -45,9 +44,11 @@ class OtelController : TelemetryController {
return NettyServerTelemetry.create(GlobalOpenTelemetry.get()).createCombinedHandler() return NettyServerTelemetry.create(GlobalOpenTelemetry.get()).createCombinedHandler()
} }
override fun startSpan(name: String): SpanHandle { override fun startSpan(command: String, key: String, dbSystem: String): SpanHandle? {
val span = tracer.spanBuilder(name) val span = tracer.spanBuilder(command)
.setSpanKind(SpanKind.CLIENT) .setSpanKind(SpanKind.CLIENT)
.setAttribute("db.system", dbSystem)
.setAttribute("db.operation", command)
.startSpan() .startSpan()
return OtelSpanHandle(span) return OtelSpanHandle(span)
} }
@@ -14,8 +14,4 @@ internal class OtelSpanHandle(
override fun setAttribute(key: String, value: Long) { override fun setAttribute(key: String, value: Long) {
delegate.setAttribute(key, value) delegate.setAttribute(key, value)
} }
override fun setAttribute(key: String, value: Boolean) {
delegate.setAttribute(key, value)
}
} }
@@ -36,6 +36,7 @@ import net.woggioni.rbcs.api.message.CacheMessage.CachePutResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueFoundResponse import net.woggioni.rbcs.api.message.CacheMessage.CacheValueFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueNotFoundResponse import net.woggioni.rbcs.api.message.CacheMessage.CacheValueNotFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent
import net.woggioni.rbcs.api.SpanHandle
import net.woggioni.rbcs.api.TelemetryController import net.woggioni.rbcs.api.TelemetryController
import net.woggioni.rbcs.common.ByteBufInputStream import net.woggioni.rbcs.common.ByteBufInputStream
import net.woggioni.rbcs.common.ByteBufOutputStream import net.woggioni.rbcs.common.ByteBufOutputStream
@@ -249,52 +250,43 @@ class RedisCacheHandler(
} }
val keyBytes = processCacheKey(msg.key, keyPrefix, digestAlgorithm) val keyBytes = processCacheKey(msg.key, keyPrefix, digestAlgorithm)
val keyString = String(keyBytes, StandardCharsets.UTF_8) val keyString = String(keyBytes, StandardCharsets.UTF_8)
val redisSpan = telemetryController?.startSpan("GET")?.apply { val redisSpan = telemetryController?.startSpan("GET", keyString, "redis")
setAttribute("db.system", "redis")
setAttribute("db.operation.name", "GET")
val remoteAddr = ctx.channel().remoteAddress()
if (remoteAddr is InetSocketAddress) {
remoteAddr.hostString?.let {
setAttribute("server.address", it)
}
setAttribute("server.port", remoteAddr.port.toLong())
}
}
val responseHandler = object : RedisResponseHandler { val responseHandler = object : RedisResponseHandler {
override fun responseReceived(response: RedisMessage) { override fun responseReceived(response: RedisMessage) {
when (response) { try {
is FullBulkStringRedisMessage -> { when (response) {
if (response === FullBulkStringRedisMessage.NULL_INSTANCE || response.content().readableBytes() == 0) { is FullBulkStringRedisMessage -> {
log.debug(ctx) { if (response === FullBulkStringRedisMessage.NULL_INSTANCE || response.content().readableBytes() == 0) {
"Cache miss for key ${msg.key} on Redis" log.debug(ctx) {
"Cache miss for key ${msg.key} on Redis"
}
sendMessageAndFlush(ctx, CacheValueNotFoundResponse(msg.key))
} else {
log.debug(ctx) {
"Cache hit for key ${msg.key} on Redis"
}
val getRequest = InProgressGetRequest(msg.key, ctx)
inProgressRequest = getRequest
getRequest.processResponse(response.content())
inProgressRequest = null
}
}
is ErrorRedisMessage -> {
val ex = RedisException("Redis error for GET ${msg.key}: ${response.content()}")
telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
}
else -> {
log.warn(ctx) {
"Unexpected response type from Redis for key ${msg.key}: ${response.javaClass.name}"
} }
telemetryController?.endSpan(redisSpan)
sendMessageAndFlush(ctx, CacheValueNotFoundResponse(msg.key)) sendMessageAndFlush(ctx, CacheValueNotFoundResponse(msg.key))
} else {
log.debug(ctx) {
"Cache hit for key ${msg.key} on Redis"
}
telemetryController?.endSpan(redisSpan)
val getRequest = InProgressGetRequest(msg.key, ctx)
inProgressRequest = getRequest
getRequest.processResponse(response.content())
inProgressRequest = null
} }
} }
} finally {
is ErrorRedisMessage -> { telemetryController?.endSpan(redisSpan)
val ex = RedisException("Redis error for GET ${msg.key}: ${response.content()}")
telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
}
else -> {
log.warn(ctx) {
"Unexpected response type from Redis for key ${msg.key}: ${response.javaClass.name}"
}
telemetryController?.endSpan(redisSpan)
sendMessageAndFlush(ctx, CacheValueNotFoundResponse(msg.key))
}
} }
} }
@@ -369,40 +361,33 @@ class RedisCacheHandler(
val expirySeconds = maxAge.toSeconds().toString() val expirySeconds = maxAge.toSeconds().toString()
val redisSpan = telemetryController?.startSpan("SET")?.apply { val redisSpan = telemetryController?.startSpan("SET", request.keyString, "redis")
setAttribute("db.system", "redis")
setAttribute("db.operation.name", "SET")
val remoteAddr = ctx.channel().remoteAddress()
if (remoteAddr is InetSocketAddress) {
remoteAddr.hostString?.let {
setAttribute("server.address", it)
}
setAttribute("server.port", remoteAddr.port.toLong())
}
}
val responseHandler = object : RedisResponseHandler { val responseHandler = object : RedisResponseHandler {
override fun responseReceived(response: RedisMessage) { override fun responseReceived(response: RedisMessage) {
when (response) { try {
is SimpleStringRedisMessage -> { when (response) {
log.debug(ctx) { is SimpleStringRedisMessage -> {
"Inserted key ${request.keyString} into Redis" log.debug(ctx) {
"Inserted key ${request.keyString} into Redis"
}
sendMessageAndFlush(ctx, CachePutResponse(request.keyString))
} }
telemetryController?.endSpan(redisSpan)
sendMessageAndFlush(ctx, CachePutResponse(request.keyString))
}
is ErrorRedisMessage -> { is ErrorRedisMessage -> {
val ex = RedisException("Redis error for SET ${request.keyString}: ${response.content()}") val ex = RedisException("Redis error for SET ${request.keyString}: ${response.content()}")
telemetryController?.endSpan(redisSpan, ex) telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex) this@RedisCacheHandler.exceptionCaught(ctx, ex)
} }
else -> { else -> {
val ex = RedisException("Unexpected response for SET ${request.keyString}: ${response.javaClass.name}") val ex = RedisException("Unexpected response for SET ${request.keyString}: ${response.javaClass.name}")
telemetryController?.endSpan(redisSpan, ex) telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex) this@RedisCacheHandler.exceptionCaught(ctx, ex)
}
} }
} finally {
telemetryController?.endSpan(redisSpan)
} }
} }
@@ -188,7 +188,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
?: return anonymousUserGroups?.let { AuthenticationResult(null, it) } ?: return anonymousUserGroups?.let { AuthenticationResult(null, it) }
val ldapName = try { val ldapName = try {
LdapName(subjectDn) LdapName(subjectDn)
} catch (_: Exception) { } catch (e: Exception) {
log.debug(ctx) { log.debug(ctx) {
"Invalid subject DN in header $headerName: $subjectDn" "Invalid subject DN in header $headerName: $subjectDn"
} }
@@ -354,7 +354,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
}?.let { }?.let {
pattern.matcher(it.value.toString()) pattern.matcher(it.value.toString())
}?.takeIf(Matcher::matches)?.group(1) }?.takeIf(Matcher::matches)?.group(1)
cfg.users[userName] ?: throw RuntimeException("Failed to extract user") cfg.users[userName] ?: throw java.lang.RuntimeException("Failed to extract user")
} }
} }
@@ -368,7 +368,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
}?.let { }?.let {
pattern.matcher(it.value.toString()) pattern.matcher(it.value.toString())
}?.takeIf(Matcher::matches)?.group(1) }?.takeIf(Matcher::matches)?.group(1)
cfg.groups[groupName] ?: throw RuntimeException("Failed to extract group") cfg.groups[groupName] ?: throw java.lang.RuntimeException("Failed to extract group")
} }
} }
@@ -344,14 +344,14 @@ object Parser {
roles = parseRoles(child) roles = parseRoles(child)
} }
"group-quota" -> { "group-quota" -> {
groupQuota = parseQuota(child) userQuota = parseQuota(child)
} }
"user-quota" -> { "user-quota" -> {
userQuota = parseQuota(child) groupQuota = parseQuota(child)
} }
} }
} }
groupName to Group(groupName, roles, groupQuota, userQuota) groupName to Group(groupName, roles, userQuota, groupQuota)
}.toMap() }.toMap()
val users = knownUsersMap.map { (name, user) -> val users = knownUsersMap.map { (name, user) ->
name to User(name, user.password, userGroups[name]?.mapNotNull { groups[it] }?.toSet() ?: emptySet(), user.quota) name to User(name, user.password, userGroups[name]?.mapNotNull { groups[it] }?.toSet() ?: emptySet(), user.quota)
@@ -23,21 +23,23 @@ class ProxyProtocolHandler(private val trustedProxyIPs : List<Cidr>) : SimpleCha
) { ) {
val sourceAddress = ctx.channel().remoteAddress() val sourceAddress = ctx.channel().remoteAddress()
if (sourceAddress is InetSocketAddress && if (sourceAddress is InetSocketAddress &&
(trustedProxyIPs.isEmpty() || trustedProxyIPs.isEmpty() ||
trustedProxyIPs.any { it.contains(sourceAddress.address) }.also { trustedProxyIPs.any { it.contains((sourceAddress as InetSocketAddress).address) }.also {
if(!it) { if(!it && log.isTraceEnabled) {
log.trace { log.trace {
"Received a proxied connection request from $sourceAddress which is not a trusted proxy address, " + "Received a proxied connection request from $sourceAddress which is not a trusted proxy address, " +
"the proxy server address will be used instead" "the proxy server address will be used instead"
} }
} }
})) { }) {
val proxiedClientAddress = InetSocketAddress( val proxiedClientAddress = InetSocketAddress(
InetAddress.ofLiteral(msg.sourceAddress()), InetAddress.ofLiteral(msg.sourceAddress()),
msg.sourcePort() msg.sourcePort()
) )
log.trace { if(log.isTraceEnabled) {
"Received proxied connection request from $sourceAddress forwarded for $proxiedClientAddress" log.trace {
"Received proxied connection request from $sourceAddress forwarded for $proxiedClientAddress"
}
} }
ctx.channel().attr(RemoteBuildCacheServer.clientIp).set(proxiedClientAddress) ctx.channel().attr(RemoteBuildCacheServer.clientIp).set(proxiedClientAddress)
} }
@@ -171,6 +171,7 @@ class ServerHandler(private val serverPrefix: Path, private val cacheHandlerSupp
ctx.pipeline().addBefore(ExceptionHandler.NAME, null, cacheHandler) ctx.pipeline().addBefore(ExceptionHandler.NAME, null, cacheHandler)
key.let(::CacheGetRequest) key.let(::CacheGetRequest)
.let(ctx::fireChannelRead) .let(ctx::fireChannelRead)
?: ctx.channel().write(CacheValueNotFoundResponse(key))
} else { } else {
cacheRequestInProgress = false cacheRequestInProgress = false
log.warn(ctx) { log.warn(ctx) {