Skip to content

Commit

Permalink
Work around Camel OpenTelemetry thread factory services not being com…
Browse files Browse the repository at this point in the history
…patible with Quarkus OpenTelemetry
  • Loading branch information
jamesnetherton committed Oct 21, 2024
1 parent 4c5b278 commit 8d54b4a
Show file tree
Hide file tree
Showing 6 changed files with 313 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,6 @@ public void bindBeansToRegistry(
})
.forEach(si -> {
LOGGER.debug("Binding bean with name: {}, type {}", si.name, si.type);

recorder.bind(
registry.getRegistry(),
si.name,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,19 @@
*/
package org.apache.camel.quarkus.component.opentelemetry.deployment;

import java.nio.file.Paths;
import java.util.Map;

import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.builditem.FeatureBuildItem;
import io.quarkus.opentelemetry.deployment.tracing.TracerEnabled;
import org.apache.camel.quarkus.component.opentelemetry.OpenTelemetryTracerProducer;
import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
import org.apache.camel.quarkus.core.deployment.spi.CamelServiceDestination;
import org.apache.camel.quarkus.core.deployment.spi.CamelServicePatternBuildItem;
import org.apache.camel.spi.FactoryFinder;

class OpenTelemetryProcessor {

Expand All @@ -38,4 +46,25 @@ AdditionalBeanBuildItem openTelemetryTracerProducerBean() {
.addBeanClass(OpenTelemetryTracerProducer.class)
.build();
}

// TODO: Remove this: https://github.com/apache/camel-quarkus/issues/6669
@BuildStep
void overrideCamelOpenTelemetryThreadPoolServices(
BuildProducer<CamelServicePatternBuildItem> camelServicePattern,
BuildProducer<CamelServiceBuildItem> camelService) {

Map.of("thread-pool-factory", "OpenTelemetryInstrumentedThreadPoolFactory",
"thread-factory-listener", "OpenTelemetryInstrumentedThreadFactoryListener")
.forEach((serviceName, type) -> {
String servicePath = FactoryFinder.DEFAULT_PATH + serviceName;
// Disable broken original service
camelServicePattern
.produce(new CamelServicePatternBuildItem(CamelServiceDestination.DISCOVERY, false, servicePath));

// Replace with working
camelService.produce(new CamelServiceBuildItem(Paths.get(servicePath),
"org.apache.camel.quarkus.component.opentelemetry.patch.%s".formatted(type)));
});

}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
/*
* 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.
*/
package org.apache.camel.quarkus.component.opentelemetry.patch;

import java.util.concurrent.ThreadFactory;

import io.opentelemetry.context.Context;
import org.apache.camel.spi.ExecutorServiceManager;
import org.apache.camel.spi.annotations.JdkService;

/**
* TODO: Remove this: https://github.com/apache/camel-quarkus/issues/6669
*/
@JdkService(ExecutorServiceManager.ThreadFactoryListener.FACTORY)
public class OpenTelemetryInstrumentedThreadFactoryListener implements ExecutorServiceManager.ThreadFactoryListener {

@Override
public ThreadFactory onNewThreadFactory(ThreadFactory factory) {
return runnable -> factory.newThread(Context.current().wrap(runnable));
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
/*
* 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.
*/
package org.apache.camel.quarkus.component.opentelemetry.patch;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import io.opentelemetry.context.Context;
import org.apache.camel.quarkus.component.opentelemetry.patch.internal.CurrentContextScheduledExecutorService;
import org.apache.camel.spi.ThreadPoolFactory;
import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.spi.annotations.JdkService;
import org.apache.camel.support.DefaultThreadPoolFactory;

/**
* TODO: Remove this: https://github.com/apache/camel-quarkus/issues/6669
*/
@JdkService(ThreadPoolFactory.FACTORY)
public class OpenTelemetryInstrumentedThreadPoolFactory extends DefaultThreadPoolFactory implements ThreadPoolFactory {

@Override
public ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return Context.taskWrapping(super.newCachedThreadPool(threadFactory));
}

@Override
public ExecutorService newThreadPool(
int corePoolSize,
int maxPoolSize,
long keepAliveTime,
TimeUnit timeUnit,
int maxQueueSize,
boolean allowCoreThreadTimeOut,
RejectedExecutionHandler rejectedExecutionHandler,
ThreadFactory threadFactory)
throws IllegalArgumentException {

ExecutorService executorService = super.newThreadPool(
corePoolSize,
maxPoolSize,
keepAliveTime,
timeUnit,
maxQueueSize,
allowCoreThreadTimeOut,
rejectedExecutionHandler,
threadFactory);

return Context.taskWrapping(executorService);
}

@Override
public ScheduledExecutorService newScheduledThreadPool(ThreadPoolProfile profile, ThreadFactory threadFactory) {
return new CurrentContextScheduledExecutorService(super.newScheduledThreadPool(profile, threadFactory));
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
/*
* 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.
*/
package org.apache.camel.quarkus.component.opentelemetry.patch.internal;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import io.opentelemetry.context.Context;

/**
* TODO: Remove this: https://github.com/apache/camel-quarkus/issues/6669
*/
public class CurrentContextScheduledExecutorService extends ForwardingScheduledExecutorService {

public CurrentContextScheduledExecutorService(ScheduledExecutorService delegate) {
super(delegate);
}

@Override
public <T> Future<T> submit(Callable<T> task) {
return delegate().submit(Context.current().wrap(task));
}

@Override
public <T> Future<T> submit(Runnable task, T result) {
return delegate().submit(Context.current().wrap(task), result);
}

@Override
public Future<?> submit(Runnable task) {
return delegate().submit(Context.current().wrap(task));
}

@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
return delegate().invokeAll(wrap(Context.current(), tasks));
}

@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
return delegate().invokeAll(wrap(Context.current(), tasks), timeout, unit);
}

@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
return delegate().invokeAny(wrap(Context.current(), tasks));
}

@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return delegate().invokeAny(wrap(Context.current(), tasks), timeout, unit);
}

@Override
public void execute(Runnable command) {
delegate().execute(Context.current().wrap(command));
}

@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
return delegate().schedule(Context.current().wrap(command), delay, unit);
}

@Override
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
return delegate().schedule(Context.current().wrap(callable), delay, unit);
}

@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
return delegate().scheduleAtFixedRate(Context.current().wrap(command), initialDelay, period, unit);
}

@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
return delegate().scheduleWithFixedDelay(Context.current().wrap(command), initialDelay, delay, unit);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
/*
* 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.
*/
package org.apache.camel.quarkus.component.opentelemetry.patch.internal;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import io.opentelemetry.context.Context;

/**
* TODO: Remove this: https://github.com/apache/camel-quarkus/issues/6669
*/
abstract class ForwardingScheduledExecutorService implements ScheduledExecutorService {

private final ScheduledExecutorService delegate;

protected ForwardingScheduledExecutorService(ScheduledExecutorService delegate) {
this.delegate = delegate;
}

ScheduledExecutorService delegate() {
return delegate;
}

@Override
public void shutdown() {
delegate.shutdown();
}

@Override
public List<Runnable> shutdownNow() {
return delegate.shutdownNow();
}

@Override
public boolean isShutdown() {
return delegate.isShutdown();
}

@Override
public boolean isTerminated() {
return delegate.isTerminated();
}

@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return delegate.awaitTermination(timeout, unit);
}

protected static <T> Collection<? extends Callable<T>> wrap(Context context, Collection<? extends Callable<T>> tasks) {
List<Callable<T>> wrapped = new ArrayList<>();
for (Callable<T> task : tasks) {
wrapped.add(context.wrap(task));
}
return wrapped;
}
}

0 comments on commit 8d54b4a

Please sign in to comment.