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
19 changed files with 94 additions and 1208 deletions
+2
View File
@@ -4,6 +4,8 @@
# Ignore Gradle build output directory
build
rbcs-cli/native-image/*.json
# Ignore JDTLS files
.classpath
.project
-5
View File
@@ -1,7 +1,6 @@
FROM eclipse-temurin:25-jre-alpine AS base-release
RUN adduser -D rbcs
USER rbcs
ENV RBCS_CONFIGURATION_DIR="/etc/rbcs"
WORKDIR /var/lib/rbcs
FROM base-release AS release-vanilla
@@ -48,12 +47,9 @@ FROM scratch AS release-native
COPY --from=base-native /etc/passwd /etc/passwd
COPY --from=base-native /etc/rbcs /etc/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 logback.xml /etc/rbcs/logback.xml
USER 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"]
FROM debian:12-slim AS release-jlink
@@ -65,5 +61,4 @@ RUN adduser -u 1000 rbcs
USER rbcs
WORKDIR /var/lib/rbcs
ADD logback.xml /etc/rbcs/logback.xml
ENV RBCS_CONFIGURATION_DIR="/etc/rbcs"
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.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
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, 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.Nullable;
import java.util.Map;
public interface TelemetryController {
void initialize();
@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);
+5 -5
View File
@@ -90,7 +90,7 @@ Provider<EnvelopeJarTask> envelopeJarTaskProvider = tasks.named(EnvelopePlugin.E
tasks.named(NativeImagePlugin.CONFIGURE_NATIVE_IMAGE_TASK_NAME, NativeImageConfigurationTask) {
toolchain {
languageVersion = JavaLanguageVersion.of(25)
vendor = JvmVendorSpec.ORACLE
vendor = JvmVendorSpec.GRAAL_VM
}
mainClass = "net.woggioni.rbcs.cli.graal.GraalNativeImageConfiguration"
classpath = project.files(
@@ -108,10 +108,10 @@ tasks.named(NativeImagePlugin.CONFIGURE_NATIVE_IMAGE_TASK_NAME, NativeImageConfi
nativeImage {
toolchain {
languageVersion = JavaLanguageVersion.of(25)
vendor = JvmVendorSpec.ORACLE
vendor = JvmVendorSpec.GRAAL_VM
}
mainClass = mainClassName
//mainModule = mainModuleName
// mainModule = mainModuleName
useMusl = true
buildStaticImage = true
linkAtBuildTime = false
@@ -119,7 +119,6 @@ nativeImage {
compressExecutable = true
compressionLevel = 6
useLZMA = false
//verbose = true
}
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) {
toolchain {
languageVersion = JavaLanguageVersion.of(25)
vendor = JvmVendorSpec.ORACLE
vendor = JvmVendorSpec.GRAAL_VM
}
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) {
exclude 'lib/libjvmcicompiler.so'
}
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 \
-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
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
#-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.atomic.AtomicInteger
import javax.net.ssl.TrustManagerFactory
import javax.net.ssl.X509ExtendedTrustManager
import javax.net.ssl.X509TrustManager
import kotlin.random.Random
import io.netty.util.concurrent.Future as NettyFuture
@@ -75,25 +74,13 @@ class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoC
)
profile.tlsTruststore?.let { trustStore ->
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?, 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?, socket: java.net.Socket) {
}
override fun checkServerTrusted(certChain: Array<out X509Certificate>, p1: String?, engine: javax.net.ssl.SSLEngine) {
}
override fun getAcceptedIssuers() = null
})
} else {
@@ -19,7 +19,6 @@ import java.security.cert.X509Certificate
import java.util.EnumSet
import java.util.ServiceLoader
import javax.net.ssl.TrustManagerFactory
import javax.net.ssl.X509ExtendedTrustManager
import javax.net.ssl.X509TrustManager
import net.woggioni.jwo.JWO
import net.woggioni.jwo.Tuple2
@@ -125,7 +124,7 @@ object RBCS {
return keystore
}
fun getTrustManager(trustStore: KeyStore?, certificateRevocationEnabled: Boolean): X509ExtendedTrustManager {
fun getTrustManager(trustStore: KeyStore?, certificateRevocationEnabled: Boolean): X509TrustManager {
return if (trustStore != null) {
val certificateFactory = CertificateFactory.getInstance("X.509")
val validator = CertPathValidator.getInstance("PKIX").apply {
@@ -137,7 +136,7 @@ object RBCS {
val params = PKIXParameters(trustStore).apply {
isRevocationEnabled = certificateRevocationEnabled
}
object : X509ExtendedTrustManager() {
object : X509TrustManager {
override fun checkClientTrusted(chain: Array<out X509Certificate>, authType: String) {
val clientCertificateChain = certificateFactory.generateCertPath(chain.toList())
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) {
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()
.filter(trustStore::isCertificateEntry)
.map(trustStore::getCertificate)
@@ -178,8 +161,8 @@ object RBCS {
}
} else {
val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
trustManagerFactory.trustManagers.asSequence().filter { it is X509ExtendedTrustManager }
.single() as X509ExtendedTrustManager
trustManagerFactory.trustManagers.asSequence().filter { it is X509TrustManager }
.single() as X509TrustManager
}
}
@@ -262,17 +262,7 @@ class MemcacheCacheHandler(
val key = ctx.alloc().buffer().also {
it.writeBytes(processCacheKey(msg.key, keyPrefix, digestAlgorithm))
}
val memcacheSpan = telemetryController?.startSpan("GET")?.apply {
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 memcacheSpan = telemetryController?.startSpan("GET", msg.key, "memcache")
val responseHandler = object : MemcacheResponseHandler {
override fun responseReceived(response: BinaryMemcacheResponse) {
val status = response.status()
@@ -328,6 +318,11 @@ class MemcacheCacheHandler(
}
}
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) {
"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"
}
putRequest.write(msg.content())
val memcacheSpan = telemetryController?.startSpan("SET",
)?.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())
}
}
val memcacheSpan = telemetryController?.startSpan("SET", putRequest.entryKey, "memcache")
putRequest.memcacheSpanRef.set(memcacheSpan)
val key = putRequest.digest.retainedDuplicate()
val (payloadSize, payloadSource) = putRequest.commit()
@@ -2,7 +2,6 @@ package net.woggioni.rbcs.server.otel
import io.netty.channel.ChannelHandler
import io.opentelemetry.api.GlobalOpenTelemetry
import io.opentelemetry.api.common.AttributeKey
import io.opentelemetry.api.trace.SpanKind
import io.opentelemetry.api.trace.StatusCode
import io.opentelemetry.instrumentation.logback.appender.v1_0.OpenTelemetryAppender
@@ -45,9 +44,11 @@ class OtelController : TelemetryController {
return NettyServerTelemetry.create(GlobalOpenTelemetry.get()).createCombinedHandler()
}
override fun startSpan(name: String): SpanHandle {
val span = tracer.spanBuilder(name)
override fun startSpan(command: String, key: String, dbSystem: String): SpanHandle? {
val span = tracer.spanBuilder(command)
.setSpanKind(SpanKind.CLIENT)
.setAttribute("db.system", dbSystem)
.setAttribute("db.operation", command)
.startSpan()
return OtelSpanHandle(span)
}
@@ -14,8 +14,4 @@ internal class OtelSpanHandle(
override fun setAttribute(key: String, value: Long) {
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.CacheValueNotFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent
import net.woggioni.rbcs.api.SpanHandle
import net.woggioni.rbcs.api.TelemetryController
import net.woggioni.rbcs.common.ByteBufInputStream
import net.woggioni.rbcs.common.ByteBufOutputStream
@@ -249,52 +250,43 @@ class RedisCacheHandler(
}
val keyBytes = processCacheKey(msg.key, keyPrefix, digestAlgorithm)
val keyString = String(keyBytes, StandardCharsets.UTF_8)
val redisSpan = telemetryController?.startSpan("GET")?.apply {
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 redisSpan = telemetryController?.startSpan("GET", keyString, "redis")
val responseHandler = object : RedisResponseHandler {
override fun responseReceived(response: RedisMessage) {
when (response) {
is FullBulkStringRedisMessage -> {
if (response === FullBulkStringRedisMessage.NULL_INSTANCE || response.content().readableBytes() == 0) {
log.debug(ctx) {
"Cache miss for key ${msg.key} on Redis"
try {
when (response) {
is FullBulkStringRedisMessage -> {
if (response === FullBulkStringRedisMessage.NULL_INSTANCE || response.content().readableBytes() == 0) {
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))
} 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
}
}
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))
}
} finally {
telemetryController?.endSpan(redisSpan)
}
}
@@ -369,40 +361,33 @@ class RedisCacheHandler(
val expirySeconds = maxAge.toSeconds().toString()
val redisSpan = telemetryController?.startSpan("SET")?.apply {
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 redisSpan = telemetryController?.startSpan("SET", request.keyString, "redis")
val responseHandler = object : RedisResponseHandler {
override fun responseReceived(response: RedisMessage) {
when (response) {
is SimpleStringRedisMessage -> {
log.debug(ctx) {
"Inserted key ${request.keyString} into Redis"
try {
when (response) {
is SimpleStringRedisMessage -> {
log.debug(ctx) {
"Inserted key ${request.keyString} into Redis"
}
sendMessageAndFlush(ctx, CachePutResponse(request.keyString))
}
telemetryController?.endSpan(redisSpan)
sendMessageAndFlush(ctx, CachePutResponse(request.keyString))
}
is ErrorRedisMessage -> {
val ex = RedisException("Redis error for SET ${request.keyString}: ${response.content()}")
telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
}
is ErrorRedisMessage -> {
val ex = RedisException("Redis error for SET ${request.keyString}: ${response.content()}")
telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
}
else -> {
val ex = RedisException("Unexpected response for SET ${request.keyString}: ${response.javaClass.name}")
telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
else -> {
val ex = RedisException("Unexpected response for SET ${request.keyString}: ${response.javaClass.name}")
telemetryController?.endSpan(redisSpan, 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) }
val ldapName = try {
LdapName(subjectDn)
} catch (_: Exception) {
} catch (e: Exception) {
log.debug(ctx) {
"Invalid subject DN in header $headerName: $subjectDn"
}
@@ -354,7 +354,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
}?.let {
pattern.matcher(it.value.toString())
}?.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 {
pattern.matcher(it.value.toString())
}?.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)
}
"group-quota" -> {
groupQuota = parseQuota(child)
userQuota = parseQuota(child)
}
"user-quota" -> {
userQuota = parseQuota(child)
groupQuota = parseQuota(child)
}
}
}
groupName to Group(groupName, roles, groupQuota, userQuota)
groupName to Group(groupName, roles, userQuota, groupQuota)
}.toMap()
val users = knownUsersMap.map { (name, user) ->
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()
if (sourceAddress is InetSocketAddress &&
(trustedProxyIPs.isEmpty() ||
trustedProxyIPs.any { it.contains(sourceAddress.address) }.also {
if(!it) {
trustedProxyIPs.isEmpty() ||
trustedProxyIPs.any { it.contains((sourceAddress as InetSocketAddress).address) }.also {
if(!it && log.isTraceEnabled) {
log.trace {
"Received a proxied connection request from $sourceAddress which is not a trusted proxy address, " +
"the proxy server address will be used instead"
}
}
})) {
}) {
val proxiedClientAddress = InetSocketAddress(
InetAddress.ofLiteral(msg.sourceAddress()),
msg.sourcePort()
)
log.trace {
"Received proxied connection request from $sourceAddress forwarded for $proxiedClientAddress"
if(log.isTraceEnabled) {
log.trace {
"Received proxied connection request from $sourceAddress forwarded for $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)
key.let(::CacheGetRequest)
.let(ctx::fireChannelRead)
?: ctx.channel().write(CacheValueNotFoundResponse(key))
} else {
cacheRequestInProgress = false
log.warn(ctx) {