Compare commits

...
3 Commits
Author SHA1 Message Date
woggioni 4b2f2da35f update to netty 4.2.15
CI / build (push) Successful in 3m52s
2026-06-12 09:43:39 +08:00
woggioni 77cc044d0e restored docker image environmental variables
CI / build (push) Successful in 3m44s
2026-05-30 12:56:03 +08:00
opencodeandwoggioni 9a7a2566fa Generalize OTEL API and add memcache tracing support
CI / build (push) Successful in 6m22s
- 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-27 23:37:50 +08:00
22 changed files with 1274 additions and 111 deletions
-2
View File
@@ -4,8 +4,6 @@
# 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
+5
View File
@@ -1,6 +1,7 @@
FROM eclipse-temurin:25-jre-alpine AS base-release FROM eclipse-temurin:25-jre-alpine AS base-release
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
@@ -47,9 +48,12 @@ FROM scratch AS release-native
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
@@ -61,4 +65,5 @@ 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.0 rbcs.version = 0.5.1
lys.version = 2026.05.16 lys.version = 2026.06.08
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
@@ -2,9 +2,12 @@ package net.woggioni.rbcs.api;
import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.NotNull;
public interface RedisSpan { 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,15 @@ 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 RedisSpan startRedisSpan(@NotNull String command, @NotNull String key); @Nullable SpanHandle startSpan(@NotNull String command);
void endRedisSpan(@Nullable RedisSpan span); void endSpan(@Nullable SpanHandle span);
void endRedisSpan(@Nullable RedisSpan span, @NotNull Throwable error); void endSpan(@Nullable SpanHandle span, @NotNull Throwable error);
} }
+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.GRAAL_VM vendor = JvmVendorSpec.ORACLE
} }
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.GRAAL_VM vendor = JvmVendorSpec.ORACLE
} }
mainClass = mainClassName mainClass = mainClassName
// mainModule = mainModuleName //mainModule = mainModuleName
useMusl = true useMusl = true
buildStaticImage = true buildStaticImage = true
linkAtBuildTime = false linkAtBuildTime = false
@@ -119,6 +119,7 @@ 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) {
@@ -127,7 +128,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.GRAAL_VM vendor = JvmVendorSpec.ORACLE
} }
mainClass = mainClassName mainClass = mainClassName
@@ -152,7 +153,6 @@ 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
+10 -1
View File
@@ -1,2 +1,11 @@
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 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
#-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,6 +10,7 @@ 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
@@ -74,13 +75,25 @@ class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoC
) )
profile.tlsTruststore?.let { trustStore -> profile.tlsTruststore?.let { trustStore ->
if (!trustStore.verifyServerCertificate) { if (!trustStore.verifyServerCertificate) {
trustManager(object : X509TrustManager { trustManager(object : X509ExtendedTrustManager() {
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,6 +19,7 @@ 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
@@ -124,7 +125,7 @@ object RBCS {
return keystore return keystore
} }
fun getTrustManager(trustStore: KeyStore?, certificateRevocationEnabled: Boolean): X509TrustManager { fun getTrustManager(trustStore: KeyStore?, certificateRevocationEnabled: Boolean): X509ExtendedTrustManager {
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 {
@@ -136,7 +137,7 @@ object RBCS {
val params = PKIXParameters(trustStore).apply { val params = PKIXParameters(trustStore).apply {
isRevocationEnabled = certificateRevocationEnabled isRevocationEnabled = certificateRevocationEnabled
} }
object : X509TrustManager { object : X509ExtendedTrustManager() {
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 {
@@ -146,10 +147,26 @@ 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)
@@ -161,8 +178,8 @@ object RBCS {
} }
} else { } else {
val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()) val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
trustManagerFactory.trustManagers.asSequence().filter { it is X509TrustManager } trustManagerFactory.trustManagers.asSequence().filter { it is X509ExtendedTrustManager }
.single() as X509TrustManager .single() as X509ExtendedTrustManager
} }
} }
@@ -1,4 +1,5 @@
import net.woggioni.rbcs.api.CacheProvider; import net.woggioni.rbcs.api.CacheProvider;
import net.woggioni.rbcs.api.TelemetryController;
module net.woggioni.rbcs.server.memcache { module net.woggioni.rbcs.server.memcache {
requires net.woggioni.rbcs.common; requires net.woggioni.rbcs.common;
@@ -16,5 +17,7 @@ module net.woggioni.rbcs.server.memcache {
provides CacheProvider with net.woggioni.rbcs.server.memcache.MemcacheCacheProvider; provides CacheProvider with net.woggioni.rbcs.server.memcache.MemcacheCacheProvider;
uses TelemetryController;
opens net.woggioni.rbcs.server.memcache.schema; opens net.woggioni.rbcs.server.memcache.schema;
} }
@@ -9,9 +9,11 @@ import java.nio.channels.FileChannel
import java.nio.channels.ReadableByteChannel import java.nio.channels.ReadableByteChannel
import java.nio.file.Files import java.nio.file.Files
import java.nio.file.StandardOpenOption import java.nio.file.StandardOpenOption
import java.net.InetSocketAddress
import java.time.Duration import java.time.Duration
import java.time.Instant import java.time.Instant
import java.util.concurrent.CompletableFuture import java.util.concurrent.CompletableFuture
import java.util.concurrent.atomic.AtomicReference
import java.util.zip.Deflater import java.util.zip.Deflater
import java.util.zip.DeflaterOutputStream import java.util.zip.DeflaterOutputStream
import java.util.zip.InflaterOutputStream import java.util.zip.InflaterOutputStream
@@ -39,8 +41,11 @@ 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.common.ByteBufInputStream import net.woggioni.rbcs.common.ByteBufInputStream
import net.woggioni.rbcs.common.ByteBufOutputStream import net.woggioni.rbcs.common.ByteBufOutputStream
import net.woggioni.rbcs.common.RBCS.loadService
import net.woggioni.rbcs.common.RBCS.processCacheKey import net.woggioni.rbcs.common.RBCS.processCacheKey
import net.woggioni.rbcs.common.RBCS.toIntOrNull import net.woggioni.rbcs.common.RBCS.toIntOrNull
import net.woggioni.rbcs.common.createLogger import net.woggioni.rbcs.common.createLogger
@@ -70,6 +75,10 @@ class MemcacheCacheHandler(
} }
} }
private val telemetryController by lazy {
loadService(TelemetryController::class.java).firstOrNull()
}
private interface InProgressRequest { private interface InProgressRequest {
} }
@@ -153,7 +162,9 @@ class MemcacheCacheHandler(
metadata: CacheValueMetadata, metadata: CacheValueMetadata,
val digest: ByteBuf, val digest: ByteBuf,
val requestController: CompletableFuture<MemcacheRequestController>, val requestController: CompletableFuture<MemcacheRequestController>,
private val alloc: ByteBufAllocator private val alloc: ByteBufAllocator,
val entryKey: String,
val memcacheSpanRef: AtomicReference<SpanHandle?>,
) : InProgressRequest { ) : InProgressRequest {
private var totalSize = 0 private var totalSize = 0
private var tmpFile: FileChannel? = null private var tmpFile: FileChannel? = null
@@ -251,6 +262,17 @@ 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 {
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()
@@ -266,8 +288,15 @@ class MemcacheCacheHandler(
log.debug(ctx) { log.debug(ctx) {
"Cache miss for key ${msg.key} on memcache" "Cache miss for key ${msg.key} on memcache"
} }
telemetryController?.endSpan(memcacheSpan)
sendMessageAndFlush(ctx, CacheValueNotFoundResponse(msg.key)) sendMessageAndFlush(ctx, CacheValueNotFoundResponse(msg.key))
} }
else -> {
val ex = MemcacheException(status)
telemetryController?.endSpan(memcacheSpan, ex)
this@MemcacheCacheHandler.exceptionCaught(ctx, ex)
}
} }
} }
@@ -282,11 +311,13 @@ class MemcacheCacheHandler(
if (content is LastMemcacheContent) { if (content is LastMemcacheContent) {
inProgressRequest = null inProgressRequest = null
inProgressGetRequest.commit() inProgressGetRequest.commit()
telemetryController?.endSpan(memcacheSpan)
} }
} }
} }
override fun exceptionCaught(ex: Throwable) { override fun exceptionCaught(ex: Throwable) {
telemetryController?.endSpan(memcacheSpan, ex)
(inProgressRequest as? InProgressGetRequest).let { inProgressGetRequest -> (inProgressRequest as? InProgressGetRequest).let { inProgressGetRequest ->
inProgressGetRequest?.let { inProgressGetRequest?.let {
inProgressRequest = null inProgressRequest = null
@@ -312,6 +343,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 memcacheSpanRef = AtomicReference<SpanHandle?>(null)
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()
@@ -320,16 +352,22 @@ class MemcacheCacheHandler(
log.debug(ctx) { log.debug(ctx) {
"Inserted key ${msg.key} into memcache" "Inserted key ${msg.key} into memcache"
} }
telemetryController?.endSpan(memcacheSpanRef.get())
sendMessageAndFlush(ctx, CachePutResponse(msg.key)) sendMessageAndFlush(ctx, CachePutResponse(msg.key))
} }
else -> this@MemcacheCacheHandler.exceptionCaught(ctx, MemcacheException(status)) else -> {
val ex = MemcacheException(status)
telemetryController?.endSpan(memcacheSpanRef.get(), ex)
this@MemcacheCacheHandler.exceptionCaught(ctx, ex)
}
} }
} }
override fun contentReceived(content: MemcacheContent) {} override fun contentReceived(content: MemcacheContent) {}
override fun exceptionCaught(ex: Throwable) { override fun exceptionCaught(ex: Throwable) {
telemetryController?.endSpan(memcacheSpanRef.get(), ex)
this@MemcacheCacheHandler.exceptionCaught(ctx, ex) this@MemcacheCacheHandler.exceptionCaught(ctx, ex)
} }
} }
@@ -339,7 +377,7 @@ class MemcacheCacheHandler(
this@MemcacheCacheHandler.exceptionCaught(ctx, ex) this@MemcacheCacheHandler.exceptionCaught(ctx, ex)
} }
} }
inProgressRequest = InProgressPutRequest(ctx.channel(), msg.metadata, key, requestController, ctx.alloc()) inProgressRequest = InProgressPutRequest(ctx.channel(), msg.metadata, key, requestController, ctx.alloc(), msg.key, memcacheSpanRef)
} }
private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) { private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
@@ -362,22 +400,41 @@ class MemcacheCacheHandler(
val request = inProgressRequest val request = inProgressRequest
when (request) { when (request) {
is InProgressPutRequest -> { is InProgressPutRequest -> {
val putRequest = request
inProgressRequest = null inProgressRequest = null
log.trace(ctx) { log.trace(ctx) {
"Received last chunk of ${msg.content().readableBytes()} bytes for memcache" "Received last chunk of ${msg.content().readableBytes()} bytes for memcache"
} }
request.write(msg.content()) putRequest.write(msg.content())
val key = request.digest.retainedDuplicate() val memcacheSpan = telemetryController?.startSpan("SET",
val (payloadSize, payloadSource) = request.commit() )?.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)
val key = putRequest.digest.retainedDuplicate()
val (payloadSize, payloadSource) = putRequest.commit()
val extras = ctx.alloc().buffer(8, 8) val extras = ctx.alloc().buffer(8, 8)
extras.writeInt(0) extras.writeInt(0)
extras.writeInt(encodeExpiry(maxAge)) extras.writeInt(encodeExpiry(maxAge))
val totalBodyLength = request.digest.readableBytes() + extras.readableBytes() + payloadSize val totalBodyLength = putRequest.digest.readableBytes() + extras.readableBytes() + payloadSize
log.trace(ctx) { log.trace(ctx) {
"Trying to send SET request to memcache" "Trying to send SET request to memcache"
} }
request.requestController.whenComplete { requestController, ex -> putRequest.requestController.whenComplete { requestController, ex ->
if (ex == null) { if (ex == null) {
val remoteAddr = requestController.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 SET request to memcache" "Sending SET request to memcache"
} }
@@ -147,6 +147,8 @@ class MemcacheClient(
channel.pipeline().addLast(handler) channel.pipeline().addLast(handler)
response.complete(object : MemcacheRequestController { response.complete(object : MemcacheRequestController {
override val channel: Channel = channel
private var channelReleased = false private var channelReleased = false
override fun sendRequest(request: BinaryMemcacheRequest) { override fun sendRequest(request: BinaryMemcacheRequest) {
@@ -1,10 +1,13 @@
package net.woggioni.rbcs.server.memcache.client package net.woggioni.rbcs.server.memcache.client
import io.netty.channel.Channel
import io.netty.handler.codec.memcache.MemcacheContent import io.netty.handler.codec.memcache.MemcacheContent
import io.netty.handler.codec.memcache.binary.BinaryMemcacheRequest import io.netty.handler.codec.memcache.binary.BinaryMemcacheRequest
interface MemcacheRequestController { interface MemcacheRequestController {
val channel: Channel
fun sendRequest(request : BinaryMemcacheRequest) fun sendRequest(request : BinaryMemcacheRequest)
fun sendContent(content : MemcacheContent) fun sendContent(content : MemcacheContent)
@@ -2,13 +2,14 @@ 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
import io.opentelemetry.instrumentation.netty.v4_1.NettyServerTelemetry import io.opentelemetry.instrumentation.netty.v4_1.NettyServerTelemetry
import io.opentelemetry.instrumentation.runtimetelemetry.RuntimeTelemetry import io.opentelemetry.instrumentation.runtimetelemetry.RuntimeTelemetry
import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk
import net.woggioni.rbcs.api.RedisSpan import net.woggioni.rbcs.api.SpanHandle
import net.woggioni.rbcs.api.TelemetryController import net.woggioni.rbcs.api.TelemetryController
import net.woggioni.rbcs.common.createLogger import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.common.info import net.woggioni.rbcs.common.info
@@ -44,21 +45,19 @@ class OtelController : TelemetryController {
return NettyServerTelemetry.create(GlobalOpenTelemetry.get()).createCombinedHandler() return NettyServerTelemetry.create(GlobalOpenTelemetry.get()).createCombinedHandler()
} }
override fun startRedisSpan(command: String, key: String): RedisSpan? { override fun startSpan(name: String): SpanHandle {
val span = tracer.spanBuilder(command) val span = tracer.spanBuilder(name)
.setSpanKind(SpanKind.CLIENT) .setSpanKind(SpanKind.CLIENT)
.setAttribute("db.system", "redis")
.setAttribute("db.operation", command)
.startSpan() .startSpan()
return RedisOtelSpan(span) return OtelSpanHandle(span)
} }
override fun endRedisSpan(span: RedisSpan?) { override fun endSpan(span: SpanHandle?) {
(span as? RedisOtelSpan)?.delegate?.end() (span as? OtelSpanHandle)?.delegate?.end()
} }
override fun endRedisSpan(span: RedisSpan?, error: Throwable) { override fun endSpan(span: SpanHandle?, error: Throwable) {
val s = (span as? RedisOtelSpan)?.delegate ?: return val s = (span as? OtelSpanHandle)?.delegate ?: return
s.recordException(error) s.recordException(error)
s.setStatus(StatusCode.ERROR) s.setStatus(StatusCode.ERROR)
s.end() s.end()
@@ -1,11 +1,11 @@
package net.woggioni.rbcs.server.otel package net.woggioni.rbcs.server.otel
import io.opentelemetry.api.trace.Span import io.opentelemetry.api.trace.Span
import net.woggioni.rbcs.api.RedisSpan import net.woggioni.rbcs.api.SpanHandle
internal class RedisOtelSpan( internal class OtelSpanHandle(
val delegate: Span, val delegate: Span,
) : RedisSpan { ) : SpanHandle {
override fun setAttribute(key: String, value: String) { override fun setAttribute(key: String, value: String) {
delegate.setAttribute(key, value) delegate.setAttribute(key, value)
@@ -14,4 +14,8 @@ internal class RedisOtelSpan(
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,7 +36,6 @@ 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.RedisSpan
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
@@ -250,48 +249,57 @@ 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?.startRedisSpan("GET", keyString) 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 responseHandler = object : RedisResponseHandler { val responseHandler = object : RedisResponseHandler {
override fun responseReceived(response: RedisMessage) { override fun responseReceived(response: RedisMessage) {
try { when (response) {
when (response) { is FullBulkStringRedisMessage -> {
is FullBulkStringRedisMessage -> { if (response === FullBulkStringRedisMessage.NULL_INSTANCE || response.content().readableBytes() == 0) {
if (response === FullBulkStringRedisMessage.NULL_INSTANCE || response.content().readableBytes() == 0) { log.debug(ctx) {
log.debug(ctx) { "Cache miss for key ${msg.key} on Redis"
"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?.endRedisSpan(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 {
telemetryController?.endRedisSpan(redisSpan) 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))
}
} }
} }
override fun exceptionCaught(ex: Throwable) { override fun exceptionCaught(ex: Throwable) {
telemetryController?.endRedisSpan(redisSpan, ex) telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex) this@RedisCacheHandler.exceptionCaught(ctx, ex)
} }
} }
@@ -361,38 +369,45 @@ class RedisCacheHandler(
val expirySeconds = maxAge.toSeconds().toString() val expirySeconds = maxAge.toSeconds().toString()
val redisSpan = telemetryController?.startRedisSpan("SET", request.keyString) 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 responseHandler = object : RedisResponseHandler { val responseHandler = object : RedisResponseHandler {
override fun responseReceived(response: RedisMessage) { override fun responseReceived(response: RedisMessage) {
try { when (response) {
when (response) { is SimpleStringRedisMessage -> {
is SimpleStringRedisMessage -> { log.debug(ctx) {
log.debug(ctx) { "Inserted key ${request.keyString} into Redis"
"Inserted key ${request.keyString} into Redis"
}
sendMessageAndFlush(ctx, CachePutResponse(request.keyString))
}
is ErrorRedisMessage -> {
val ex = RedisException("Redis error for SET ${request.keyString}: ${response.content()}")
telemetryController?.endRedisSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
}
else -> {
val ex = RedisException("Unexpected response for SET ${request.keyString}: ${response.javaClass.name}")
telemetryController?.endRedisSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
} }
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)
}
else -> {
val ex = RedisException("Unexpected response for SET ${request.keyString}: ${response.javaClass.name}")
telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex)
} }
} finally {
telemetryController?.endRedisSpan(redisSpan)
} }
} }
override fun exceptionCaught(ex: Throwable) { override fun exceptionCaught(ex: Throwable) {
telemetryController?.endRedisSpan(redisSpan, ex) telemetryController?.endSpan(redisSpan, ex)
this@RedisCacheHandler.exceptionCaught(ctx, ex) this@RedisCacheHandler.exceptionCaught(ctx, ex)
} }
} }
@@ -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 (e: Exception) { } catch (_: 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 java.lang.RuntimeException("Failed to extract user") cfg.users[userName] ?: throw 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 java.lang.RuntimeException("Failed to extract group") cfg.groups[groupName] ?: throw RuntimeException("Failed to extract group")
} }
} }
@@ -344,14 +344,14 @@ object Parser {
roles = parseRoles(child) roles = parseRoles(child)
} }
"group-quota" -> { "group-quota" -> {
userQuota = parseQuota(child) groupQuota = parseQuota(child)
} }
"user-quota" -> { "user-quota" -> {
groupQuota = parseQuota(child) userQuota = parseQuota(child)
} }
} }
} }
groupName to Group(groupName, roles, userQuota, groupQuota) groupName to Group(groupName, roles, groupQuota, userQuota)
}.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,23 +23,21 @@ 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 as InetSocketAddress).address) }.also { trustedProxyIPs.any { it.contains(sourceAddress.address) }.also {
if(!it && log.isTraceEnabled) { if(!it) {
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()
) )
if(log.isTraceEnabled) { log.trace {
log.trace { "Received proxied connection request from $sourceAddress forwarded for $proxiedClientAddress"
"Received proxied connection request from $sourceAddress forwarded for $proxiedClientAddress"
}
} }
ctx.channel().attr(RemoteBuildCacheServer.clientIp).set(proxiedClientAddress) ctx.channel().attr(RemoteBuildCacheServer.clientIp).set(proxiedClientAddress)
} }
@@ -171,7 +171,6 @@ 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) {