RSS

Openshift Fuse health checks with Jolokia

I recently ran into a problem where I needed to create an Openshift healthcheck for a Fuse container. Normally all Fuse containers exposed an http endpoint which was used in the healthcheck. However additional security requirements dictated the use of client certificates. Currently it is not possible to create a healthcheck with a two-way-ssl connection.

As another way to monitor if all Camel routes are started I decided to leverage the JMX beans exposed by Jolokia. For those who are unfamiliar with Jolokia, it essentially is JMX over json/http.

 

The problem was that by default the Jolokia endpoint is secured with basic authentication, and the password is generated for each created container in Openshift.

However, the Jolokia password for each container is available in the base image (FIS 2 fis-java base image was used) of each container: /opt/etc/jolokia.pw

This file can be used to execute the curl request to the Jolokia endpoint:

curl -k -v https://jolokia:`cat /opt/jolokia/etc/jolokia.pw`@localhost:8778/jolokia/exec/org.apache.camel:context=myCamelContext,type=routes,name=%22myJettyEndpoint%22/getState%28%29

In the above example the state of the “myJettyEndpoint” camel endpoint is requested.

Advertisements
 
Leave a comment

Posted by on 2017-11-24 in JBoss Fuse, Openshift

 

Tags: , , , ,

Openshift limits explained

Openshift is a Paas platform offered by Red Hat based mainly on Docker and Kubernetes. One of the concepts behind it is that Ops can set boundaries for Dev. For example by providing a list of supported technologies in the form of base images.
One other way Ops can further control the Paas cluster is to impose various limits on the components running in Openshift.

However, Openshift currently has three different ways of setting restrictions on different levels which do interconnect in an implicit way. Which sometimes can become difficult to setup in a proper way and people end up with Pods never leaving the “Pending” state. So in this blogpost we are going to take look at the different limits or restrictions available in Openshift and how they influence each other.

But to understand the restrictions better it is good to know some basic Openshift concepts and components on which these limits act. Although I highly recommend to start experimenting with restrictions and limits after you become familiar with Openshift.

Below are the components of Openshift influenced by the restrictions. (source: https://docs.openshift.com/enterprise/3.0/architecture/core_concepts/index.html)

Containers

The basic units of OpenShift applications are called containers. Linux container technologies are lightweight mechanisms for isolating running processes so that they are limited to interacting with only their designated resources. Many application instances can be running in containers on a single host without visibility into each others’ processes, files, network, and so on. Typically, each container provides a single service (often called a “micro-service”), such as a web server or a database, though containers can be used for arbitrary workloads.

The Linux kernel has been incorporating capabilities for container technologies for years. More recently the Docker project has developed a convenient management interface for Linux containers on a host. OpenShift and Kubernetes add the ability to orchestrate Docker containers across multi-host installations.

Pods

OpenShift leverages the Kubernetes concept of a pod, which is one or more containers deployed together on one host, and the smallest compute unit that can be defined, deployed, and managed.

Namespaces

A Kubernetes namespace provides a mechanism to scope resources in a cluster. In OpenShift, a project is a Kubernetes namespace with additional annotations.

Namespaces provide a unique scope for:

  • Named resources to avoid basic naming collisions.
  • Delegated management authority to trusted users.
  • The ability to limit community resource consumption.

Most objects in the system are scoped by namespace, but some are excepted and have no namespace, including nodes and users.

The Kubernetes documentation has more information on namespaces.

 

Openshift limits and restrictions

There are three different types of limits and restrictions available in Openshift.

  • Quotas
  • Limit ranges
  • Compute resources

Quotas

Quotas are boundaries configured per namespace and act as a upper limit for resources in that particular namespace. It essentially defines the capacity of the namespace. How this capacity is used is up to the user using the namespace. For example if the total capacity us used by one or one hundred pods is not dictated by the quota, except when a max number of pods is configured.

Like most things in Openshift you can configure a quota with a yaml configuration. One basic configuration for a quota looks like:

apiVersion: v1
kind: ResourceQuota
metadata:
  name: namespace-quota
spec:
  hard:
    pods: "5" 
    requests.cpu: "500m" 
    requests.memory: 512Mi 
    limits.cpu: "2" 
    limits.memory: 2Gi 

This quota says that the namespace can have a maximum of 5 pods, and/or a max of 2 cores and 2 Gb of memory, the initial “claim” of this namespace is 500 millicores and 512 Mb of memory.

Now it is important to note that by default these limits are imposed to “NonTerminating” pods only. Meaning that for example build pods which terminate eventually are not counted in this quota.

Explicitly this can be configured by adding a scope to the yaml spec:


apiVersion: v1
kind: ResourceQuota
metadata:
  name: namespace-quota
spec:
  hard:
    pods: "5" 
    requests.cpu: "500m" 
    requests.memory: 512Mi 
    limits.cpu: "2" 
    limits.memory: 2Gi 
scopes:
- NotTerminating

There are also other scopes available, like Best effort and guaranteed.

Limit ranges

One other type of limit is the “limit range”. A limit range is also configured on a namespace, however a limit range defines limits per pod and/or container in that namespace. It essentially provides CPU and memory limits for containers and pods.

Again, configuring a limit range is also done by a yaml configuration:


apiVersion: "v1"
kind: "LimitRange"
metadata:
  name: "resource-limits" 
spec:
  limits:
    -
      type: "Pod"
      max:
        cpu: "2" 
        memory: "1Gi" 
      min:
        cpu: "200m" 
        memory: "6Mi" 
    -
      type: "Container"
      max:
        cpu: "2" 
        memory: "1Gi" 
      min:
        cpu: "100m" 
        memory: "4Mi" 
      default:
        cpu: "300m" 
        memory: "200Mi" 
      defaultRequest:
        cpu: "200m" 
        memory: "100Mi"

Here we can see both Pod and Container limits. These limits define the “range” (hence the term limit range) for each container of pod in the namespace. So in the above example each Pod in the namespace will initially claim 200 millicores and 6Mb of memory and can run with a max of 1 GB of memory and 2 cores of CPU. The actual limits the Pod or container runs with can be defined in the Pod or Container spec which we will discover below. However the limit range defines the range of these limits.

Another thing to note is the default and defaultRequest limits in the Container limit range. These are the limits applied to a container who do not specify further limits and hence get assigned the default.

Compute resources

The last of the limits is probably the easiest to understand, compute resources are defined on the Pod or the Container spec, in for example the deploymentconfig or the replication controller. And define the CPI and Memory limits for that particular pod.

Lets look at an example Yaml:

 


apiVersion: v1
kind: Pod
spec:
  containers:
  - image: nginx
    name: nginx
    resources:
      requests:
        cpu: 100m 
        memory: 200Mi 
      limits:
        cpu: 200m 
        memory: 400Mi 

In the above spec the Pod will initially claim 100 millicores and 200 Mb of memory and will max out at 200 millicores and 400 Mb of memory. Note whenever a Limit range is also provided in the namespace where the above Pod runs and the compute resources limits here are within the limit range the Pod will run fine. If however the limits are above the limits in the limit range the pod will not start.

Scopes

All limits have a request and a max which define further ranges the Pod can operate on. Where the request is by all intense and purposes “guaranteed” (as long as the underlying node has the capacity). This gives the option to implicitly set different QoS tiers.

  1. BestEffort – no limits are provided whatsoever. The Pod claims whatever it needs, but is the first one to get scaled down or killed when other Pods request for resources.
  2. Burstable – The request limits are lower than the max limits. The initial limit is guaranteed, but the Pod can, if resources are available, burst to its maximum.
  3. Guaranteed – the request and max are identical, so it directly claims the max resources, even though the pod might not initially use all resources they are already claimed by the cluster, and therefore guaranteed.

Below is an overall view of the three different Openshift limits.

 

 
Leave a comment

Posted by on 2017-08-25 in Openshift

 

Tags: , , , , , ,

AES-256 message encryption in Apache Camel

This blog post shows how to encrypt and decrypt the payload of the message using Apache Camel. The cryptografic algorithm used in this example is AES-256 since this was an explicit request from security. The key used in the example was obtained from a keystore.

For extra security purposes AES encryption can be extended by using a so called Initialization Vector, which is similar as a NONCE a random number used per request. In this example a random 16 bit byte[] is used.

For more information about AES encryption: https://en.wikipedia.org/wiki/Advanced_Encryption_Standard

For more information about Initialization Vectors: https://en.wikipedia.org/wiki/Initialization_vector

To encrypt and decrypt messages Camel has made an abstraction independend of the algorithm used for encryption and decryption. This abstraction is called the CryptoDataFormat and can be used as any other data format. The CryptoDataFormat is well documented here: http://camel.apache.org/crypto.html However, the use with the AES-256 encryption algorithm is less well documented so, hopefully this post helps someone.

 

Generating the key and the keystore

The first step is to generate the key we are going to use for encryption AND decryption (remember this is symmetric encryption, meaning the same key is used for encryption als for decryption opposed to PKI which is an assymmetric encryption technique).

For generating the key we can use keytool. The example below I conviently borrowed from this blogpost: https://dzone.com/articles/aes-256-encryption-java-and

keytool -genseckey -keystore aes-keystore.jck -storetype jceks -storepass mystorepass -keyalg AES -keysize 256 -alias jceksaes -keypass mykeypass

To retrieve the key from the keystore I created some helper method, nothing special so far.

public static Key getKeyFromKeystore(KeyConfig keyConfig) {

  String keystorePass = keyConfig.getKeystorePass();
  String alias = keyConfig.getAlias();
  String keyPass = keyConfig.getKeyPass();
  String keystoreLocation = keyConfig.getKeystoreLocation();
  String keystoreType = keyConfig.getKeystoreType();

  InputStream keystoreStream = null;
  KeyStore keystore = null;
  Key key = null ;

  try {
    keystoreStream = new FileInputStream(keystoreLocation);
    keystore = KeyStore.getInstance(keystoreType);

    keystore.load(keystoreStream, keystorePass.toCharArray());
    if (!keystore.containsAlias(alias)) {
      throw new RuntimeException("Alias for key not found");
    }

    key = keystore.getKey(alias, keyPass.toCharArray());

  } catch (Exception e) {
    e.printStackTrace();
  }

  return key;
}

The KeyConfig object is just a POJO containing some String values retrieved from a propertyfile.

Since we are going to use an Initialization Vector for extra security we create a helper method for this as well, which returns a 16 bit byte array:

public static byte[] generateIV() {
  byte[] iv = new byte[16];
  new Random().nextBytes(iv);

  return iv;
}

With the helper methods in place we can turn our attention to some Camel code…

Encrypting and decrypting using a Camel Route

The first thing we need is a CryptoDataFormat. Since we are using CDI for the bootstrapping we are going to use a PostConstruct annotated method to create a CryptoDataFormat. The trick is to set the encryption algorithm to “AES/CBC/PKCS5Padding” and enter a AES-256 key.

@PostConstruct
public void setupEncryption() {
  cryptoFormat = new CryptoDataFormat("AES/CBC/PKCS5Padding", EncryptionUtils.getKeyFromKeystore(keyConfig), "SunJCE");
}

When using a CryptoDataFormat in Apache Camel we can simply use the marshal and unmarshal statements within the Camel route. However, since we are creating a message specific Initialization Vector we need to set in as well. For this we can use the helper method we created earlier. Another way would be to use a proccessor.

.setHeader("iv", method("nl.rubix.eos.poc.util.EncryptionUtils", "generateIV"))
.bean(cryptoFormat, "setInitializationVector(${header.iv})")
.marshal(cryptoFormat)

Since the example uses the same instance of the CryptoDataFormat decryption is as simple as:

.unmarshal(cryptoFormat)

In practice it won’t often be feasible to use the same intance of the CryptoDataFormat for both encryption and decrypion. When decrypting the exact same key and Initialization Vector must be used to initialize the CryptoDataFormat instance used for decryption as was used for encryption. Since this is symmetric encryption. The key used for encrypion must be available to the party which performs the decryption. This opposed to assymmetric public/private key signage. This is inherently part of the AES encryption protocol.

The complete code repo can be found here: 

java.security.InvalidKeyException: Illegal key size

For reasons unknown to mankind Oracle decided not to include key sizes of 256 in the standard JRE despite the fact we are living in the 21st century. This results in a Caused by: java.security.InvalidKeyException: Illegal key size exception. To resolve this the so called “Unlimited Strength Juristriction Policy” files must be downloaded and extracted to the JRE.

Since Oracle has a tendency to change download links it won’t be posted here. But searching the internet will provide plenty of examples on how to download and install.

 
Leave a comment

Posted by on 2017-07-10 in JBoss Fuse

 

Tags: , , , , , ,

Camel Split using a custom Iterator

One of the more commonly used EIP’s in Camel is the Splitter, you can find the documentation here http://camel.apache.org/splitter.html

Usually the splitter is used for tokenizing some message or splitting collections into single messages. But what if you need something more specific? It is good to know that the splitter can either split on Java collection types as well as Iterators. In this blog we are going to create a custom Iterator helper class and method. The business use of this example will be rather questionable, but since we have the opportunity to fully integrate a custom Java method the potential is limitless.

The first thing to do is create a Class and method which return an Iterator.

import javax.inject.Named;
import java.util.Iterator;
import java.util.NoSuchElementException;

/**
 * Created by pim on 6/30/17.
 */
@Named("myCustomIterator")
public class MyCustomIterator {

    public Iterator getIterator(Exchange exchange) {
        final String exampleMsg = exchange.getIn().getBody(String.class);

        if (exampleMsg == null)
            throw new NullPointerException();

        return new Iterator<Character>() {
            private int index = 0;

            public boolean hasNext() {
                return index < exampleMsg.length();
            }

            public Character next() {

                if (!hasNext())
                    throw new NoSuchElementException();
                return exampleMsg.charAt(index++);
            }

            public void remove() {
                throw new UnsupportedOperationException();
            }
        };

    }
}

Once we have our custom Iterator we can incorporate it in our Camel route and more specifically use it in our splitter.

package nl.rubix.eos.camel;

import javax.inject.Inject;
import javax.inject.Named;

import org.apache.camel.Endpoint;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.cdi.ContextName;
import org.apache.camel.cdi.Uri;

/**
 * Configures all our Camel routes, components, endpoints and beans
 */
@ContextName("myIteratorSplitter")
public class MyIteratorSplitterRoute extends RouteBuilder {

    @Inject
    @Named("myCustomIterator")
    MyCustomIterator myCustomIterator;

    @Override
    public void configure() throws Exception {
        // you can configure the route rule with Java DSL here

        from("direct:start")
            .log("${body}")
            .split().method("myCustomIterator", "getIterator")
                .log("${body}")
            .end();
    }

}

Now when we run this example with the string “camelisawesome” we see the following log entries:


2017-06-30 14:18:35,830 [main ] INFO DefaultCamelContext - Route: route1 started and consuming from: Endpoint[direct://start]
2017-06-30 14:18:35,830 [main ] INFO DefaultCamelContext - Total 1 routes, of which 1 are started.
2017-06-30 14:18:35,835 [main ] INFO DefaultCamelContext - Apache Camel 2.17.0.redhat-630187 (CamelContext: myIteratorSplitter) started in 0.329 seconds
2017-06-30 14:18:35,849 [main ] INFO Bootstrap - WELD-ENV-002003: Weld SE container camel-context-cdi initialized
2017-06-30 14:18:35,877 [main ] INFO route1 - camelisawesome
2017-06-30 14:18:35,887 [main ] INFO route1 - c
2017-06-30 14:18:35,887 [main ] INFO route1 - a
2017-06-30 14:18:35,888 [main ] INFO route1 - m
2017-06-30 14:18:35,888 [main ] INFO route1 - e
2017-06-30 14:18:35,889 [main ] INFO route1 - l
2017-06-30 14:18:35,889 [main ] INFO route1 - i
2017-06-30 14:18:35,889 [main ] INFO route1 - s
2017-06-30 14:18:35,890 [main ] INFO route1 - a
2017-06-30 14:18:35,890 [main ] INFO route1 - w
2017-06-30 14:18:35,891 [main ] INFO route1 - e
2017-06-30 14:18:35,891 [main ] INFO route1 - s
2017-06-30 14:18:35,891 [main ] INFO route1 - o
2017-06-30 14:18:35,892 [main ] INFO route1 - m
2017-06-30 14:18:35,893 [main ] INFO route1 - e
2017-06-30 14:18:35,898 [main ] INFO CamelContextProducer - Camel CDI is stopping Camel context [myIteratorSplitter]

Using the custom Iterator enables you to create a custom splitter.

 
Leave a comment

Posted by on 2017-06-30 in JBoss Fuse

 

Tags: , , ,

Playing around with Camel AsyncProcessor

One of the most frequently used constructs in Apache Camel is the Processor (http://camel.apache.org/processor.html), it is used ofter for invoking custom code or performing message translations. The API of the processor is very clear and well documented. As are numerous examples available for using a Camel Processor. The lesser known brother of the Processor is the AsyncProcessor (http://camel.apache.org/asynchronous-processing.html) which is less documented and a less frequently used. Mainly because the AsyncProcessor is mainly targeted for Camel Component developers. However, recently I decided to play around with the Camel AsyncProcessor in a regular Camel setup. In this blog I would like to explain one possible way how to use the AsyncProcessor in a Camel route setup.

Creating a AsyncProcessor

Similar to creating a Processor, creating a AsyncProcessor starts by implementing the AsyncProcessor interface.


public class MyAsyncProcessor implements AsyncProcessor {

But instead of one process method, now two must be implemented:


@Override public boolean process(final Exchange exchange, AsyncCallback asyncCallback) {

@Override public void process(Exchange exchange) {

Next to the regular process method  process method which takes an AsyncCallback must be implemented. The AsyncCallback is invoked whenever the Async execution (which must be started in a seperate thread) is finished. The return boolean indicates whether or not the Camel routing engine must wait or continue routing to other components/processors defined in the Camel route.

Starting the Async job

In order to start a async job the execution must take place in another thread. Java nowadays has multiple ways for concurrent, multi threaded execution. For this example we simply create a Runnable class en start the job via een executor service.


private class AsyncBackgroundProcess implements Runnable {
  private Exchange exchange;
  private AsyncCallback asyncCallback;
  public AsyncBackgroundProcess(Exchange exchange, AsyncCallback asyncCallback){
    this.exchange = exchange;
    this.asyncCallback = asyncCallback;
  }
  @Override public void run() {
    log.info("Async backend process started");
    Boolean getBoolean = slowExecutionInterface.getBoolean("bla");
    exchange.setProperty("Response", getBoolean);
    log.info("Async backend process completed");
    asyncCallback.done(false);
  }
}

There are two things to note here:

  1. The AsyncCallback object from the process method is passed to the Runnable class
  2. When the execution is finished the asyncCallback.done method is called
    1. The false parameter indicates if the execution is handled synchronously (true) or asynchronously (false)

Implementing the AsyncProcessor process method

In the Async Processor process method we kickoff the Runnable class and define the callback method:


@Override public boolean process(final Exchange exchange, AsyncCallback asyncCallback) {
  log.info("Async process started");
  CountDownLatch countDownLatch = new CountDownLatch(1);
  exchange.setProperty("countDownLatch", countDownLatch);
  executorService.submit(new AsyncBackgroundProcess(exchange, new AsyncCallback() {
    @Override public void done(boolean b) {
      log.info("Async backend process fininshed");
      exchange.getContext().getAsyncProcessorAwaitManager().countDown(exchange, exchange.getProperty("countDownLatch", CountDownLatch.class));
    }
  }));
  return true;
}

Using the AsyncProcessor in a Camel route

Using a AsyncProcessor in a Camel route is exactly the same as using a regular Processor in the route:


from(jettyEndpoint)
.log("received")
.process(myAsyncProcessor)

Getting a response

By default the AsyncProcessor triggers a new thread for execution and does not sync back to the main execution thread. So getting a response in a “Fork-Join” manner requires some additional work. In the implementation of the process method above some actions for getting a response are already present:

A CountDownLatch is used for defining the number of threads the main thread can wait for (in our case 1):


CountDownLatch countDownLatch = new CountDownLatch(1);

in order to use the CountDownLatch downstream in our Camel route we save the it to an exchange property:


exchange.setProperty("countDownLatch", countDownLatch);

in the trigger for the Runnable class in a new thread we define the AsyncCallback and its actions when the done method is invoked, to count down the CountDownLatch indicating the background thread is finished:


executorService.submit(new AsyncBackgroundProcess(exchange, new AsyncCallback() {
    @Override public void done(boolean b) {
      log.info("Async backend process fininshed");
      exchange.getContext().getAsyncProcessorAwaitManager().countDown(exchange, exchange.getProperty("countDownLatch", CountDownLatch.class));
    }
  }));

But this does not synchronize our threads just that. In order to get a response in our Camel route or Processor the execution must wait at some point in time for the background thread to complete. For this a helper method was created:


public static void getResponseInBody(Exchange exchange) {
  CountDownLatch countDownLatch = exchange.getProperty("countDownLatch", CountDownLatch.class);
  exchange.getContext().getAsyncProcessorAwaitManager().await(exchange, countDownLatch);
  log.debug("Retrieved async response " + exchange.getProperty("Response", String.class));
  exchange.getIn().setBody(exchange.getProperty("Response", String.class));
}

In this helper method the CountDownLatch is used from the exchange and the Camel AwaitManager is used for the thread synchronization:

CountDownLatch countDownLatch = exchange.getProperty("countDownLatch", CountDownLatch.class);
exchange.getContext().getAsyncProcessorAwaitManager().await(exchange, countDownLatch);

Since this helper method is static, it can be invoked from anywhere in the route or processor, thereby giving the flexibility where in the route the threads must be synchronized.


@Override
public void configure() throws Exception {
from(jettyEndpoint)
.log("received")
.process(myAsyncProcessor)
.log("exited processor")
.bean(MyAsyncProcessor.class, "getResponse(${exchange})")
.log("${body} and Response Property ${property.Response}");
}

The entire AsyncProcessor looks like this:


package nl.rubix.eos.poc.asyncprocessor;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.Exchange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.inject.Inject;
import javax.inject.Named;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Named("myAsyncProcessor")
public class MyAsyncProcessor implements AsyncProcessor {
  private final ExecutorService executorService = Executors.newFixedThreadPool(2);
  private static Logger log = LoggerFactory.getLogger(MyAsyncProcessor.class);
  @Inject
  @Named("slowExecutionImpl")
  private SlowExecutionInterface slowExecutionInterface;
  @Override public boolean process(final Exchange exchange, AsyncCallback asyncCallback) {
    log.info("Async process started");
    CountDownLatch countDownLatch = new CountDownLatch(1);
    exchange.setProperty("countDownLatch", countDownLatch);
    executorService.submit(new AsyncBackgroundProcess(exchange, new AsyncCallback() {
      @Override public void done(boolean b) {
        log.info("Async backend process fininshed");
        exchange.getContext().getAsyncProcessorAwaitManager().countDown(exchange, exchange.getProperty("countDownLatch", CountDownLatch.class));
      }
    }));
    return true;
  }
  @Override public void process(Exchange exchange) throws Exception {
    throw new IllegalStateException("Should never be called");
  }
  private class AsyncBackgroundProcess implements Runnable {
    private Exchange exchange;
    private AsyncCallback asyncCallback;
    public AsyncBackgroundProcess(Exchange exchange, AsyncCallback asyncCallback){
      this.exchange = exchange;
      this.asyncCallback = asyncCallback;
    }
    @Override public void run() {
      log.info("Async backend process started");
      Boolean getBoolean = slowExecutionInterface.getBoolean("bla");
      exchange.setProperty("Response", getBoolean);
      log.info("Async backend process completed");
      asyncCallback.done(false);
    }
  }
  public static void getResponseInBody(Exchange exchange) {
    CountDownLatch countDownLatch = exchange.getProperty("countDownLatch", CountDownLatch.class);
    exchange.getContext().getAsyncProcessorAwaitManager().await(exchange, countDownLatch);
    log.debug("Retrieved async response " + exchange.getProperty("Response", String.class));
    exchange.getIn().setBody(exchange.getProperty("Response", String.class));
  }
  public static void getResponseInProperty(Exchange exchange) {
    CountDownLatch countDownLatch = exchange.getProperty("countDownLatch", CountDownLatch.class);
    exchange.getContext().getAsyncProcessorAwaitManager().await(exchange, countDownLatch);
    log.debug("Retrieved async response " + exchange.getProperty("Response", String.class));
  }
  public static Boolean getResponse(Exchange exchange) {
    CountDownLatch countDownLatch = exchange.getProperty("countDownLatch", CountDownLatch.class);
    exchange.getContext().getAsyncProcessorAwaitManager().await(exchange, countDownLatch);
    log.debug("Retrieved async response " + exchange.getProperty("Response", String.class));
    return exchange.getProperty("Response", Boolean.class);
  }
}

The entire Camel route looks like this:


package nl.rubix.api.poc;
import nl.rubix.eos.poc.asyncprocessor.MyAsyncProcessor;
import org.apache.camel.Endpoint;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.cdi.ContextName;
import org.apache.camel.cdi.Uri;
import javax.inject.Inject;
import javax.inject.Named;
/**
 * Configures all our Camel routes, components, endpoints and beans
 */
@ContextName("myJettyCamel")
public class MyJettyRoute extends RouteBuilder {
    @Inject @Uri("jetty:http://0.0.0.0:8080/async/test")
    private Endpoint jettyEndpoint;
    @Inject
    @Named("myAsyncProcessor")
    MyAsyncProcessor myAsyncProcessor;
    @Override
    public void configure() throws Exception {
        from(jettyEndpoint)
            .log("received")
            .process(myAsyncProcessor)
            .log("exited processor")
            .bean(MyAsyncProcessor.class, "getResponse(${exchange})")
            .log("${body} and Response Property ${property.Response}");
    }
}

To simulate slow execution a simple interface and corresponding implementation where used:


package nl.rubix.eos.poc.asyncprocessor;
public interface SlowExecutionInterface {
  Boolean getBoolean(String input);
}

package nl.rubix.eos.poc.asyncprocessor;
import javax.inject.Named;
@Named("slowExecutionImpl")
public class SlowExecutionInterfaceMock implements SlowExecutionInterface {
  @Override public Boolean getBoolean(String input) {
    try {
      Thread.sleep(5000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return true;
  }
}

 
Leave a comment

Posted by on 2017-05-26 in JBoss Fuse

 

Tags: , , , , ,

Apache Camel – Dynamic redelivery based on MEP

The exception handling and retry mechanisms in Apache Camel are quite extensive. In this blogpost we are going to take a look at customizing the retry based on a predicate implementation of our own thereby enabling really fine grained retry logic.
In our example we are going to look at the MEP of the exchange, whenever it is inOnly we are going to retry since no synchronous subscriber is waiting for the response. Conversely, when the MEP of the exchange is inOut we are not going to retry and immediately send back an error. Think fail fast en stuff.
Like mentioned above we can enable this by implementing the Predicate interface from Apache Camel.
The entire class looks like this:
package nl.schiphol.api.integration.redelivery;

import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Predicate;
import org.apache.camel.component.properties.PropertiesComponent;

import javax.inject.Inject;
import javax.inject.Named;

@Named
public class InvokeEndpointsRedeliveryPolicy implements Predicate{

 @Inject
 PropertiesComponent properties;

 private Integer MAX_REDELIVERIES;

 /**
 * Returns a Boolean whether or not a redelivery must be executed.
 * Whether or not retries should be executed are dependend on two criteria: the MEP of the exchange and the max retries
 * If the MEP is inOnly a response is not required immediately and retries can commense up until the max retries
 * @implements org.apache.camel.Predicate
 * @param exchange The Camel Exchange is used as a parameter, this is dictated by the Predicate interface
 * @return Boolean if a redelivery should be executed.
 */
 @Override
 public boolean matches(Exchange exchange) {
   setMaxRedeliveries();
   //no redelivery is the default
   Boolean shouldRedeliver = false;

   ExchangePattern exchangePattern = exchange.getPattern();
   Boolean isInOut = exchangePattern.isOutCapable();
   Integer redeliveryCounter = getRedeliveryCounter(exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER, Integer.class));

   if(!isInOut && redeliveryCounter < MAX_REDELIVERIES){
    shouldRedeliver = true;
   }

   return shouldRedeliver;
 }

 private void setMaxRedeliveries() {
   try {
   String retryAmountProperty = properties.parseUri("{{health-api.invokeEndpoints.retryAmount}}");
   MAX_REDELIVERIES = new Integer(retryAmountProperty);
   } catch (Exception e) {
     MAX_REDELIVERIES = 3;
   }
 }

 private Integer getRedeliveryCounter(Integer redeliveryCounterHeader){
   Integer redeliveryCounter = redeliveryCounterHeader;

   if(redeliveryCounter == null){
    redeliveryCounter = 0;
   }

   return redeliveryCounter;
 }

 private Boolean getRetryProperty(Boolean retryProperty) {
   Boolean retry = retryProperty;

   if(retry == null){
     retry = false;
   }

   return retry;
 }
}
In order to use our predicate in our Camel Route for determining the retry use the ‘retryWhile’ statement:
@Override
private void configure() {
 from("direct:start").routeId("my-dynamic-retry-route")
 .onException(Exception.class).redeliveryDelay(1500).retryWhile(redeliveryPolicy).continued(true).end()
 ...
 
Leave a comment

Posted by on 2017-04-06 in JBoss Fuse

 

Tags: , , , , , , ,

Creating an insecure http4 component in Apache Camel

Recently I was struggling with invoking HTTP endpoints using self-signed certificates using the Apache Camel http4 component. The crux of the problem was the fact these certificates change rapidly and are maintained by other teams. Since this was an internal call only, routed through a VPN I decided  to approach the problem by disabling the certificate check instead of adding the self-signed certificates to a keystore which I normally do in these situations.  So keep in mind, doing this will undermine the security of the TLS connection. And obviously only works with one-way TLS connections.
It was with disabling the certificate checks in the http4 component that I was struggling so I decided to share what I did with you when I eventually found out the solution.
A small note, the example provided in this blogpost is using Apache Camel in combination of CDI. But the solution will work equally well with frameworks like Spring or OSGi Blueprint.
First I created my own instance of the HttpComponent seperating it from the standard http4 component.
public class CustomHttp4Component {

    @Produces
    @Named("insecurehttps4")
    public HttpComponent createCustomHttp4Component() {
       HttpComponent httpComponent = new HttpComponent();

In this custom http4 component we have to do two things. First is to change the Certificate HostnameVerifier so the hostname on the self-signed certificate is not causing an exception.

This is accomplished by setting an instance of the org.apache.http.conn.ssl.AllowAllHostnameVerifier to the http4 component using the setter method setX509HostNameVerifier. Note, the version of Apache Camel I am using (2.17) still requires the now deprecated: org.apache.http.conn.ssl.AllowAllHostnameVerifier newer versions of Apache Camel are using the org.apache.http.conn.ssl.NoopHostnameVerifier.

This unfortunately was not enough to invoke the endpoint, an empty X509TrustManager was also required. It needs to be empty for our purpose to basically omit the certificate validation checks. For this we needed to extend the X509ExtendedTrustManager and override the methods implementing them as “empty”. To set our empty thrustmanager on the http4 component we need to wrap it in a TrustManagersParameters class and wrap this into an SSLParameters class before we can add it to the http4 component.



TrustManagersParameters trustManagersParameters = new TrustManagersParameters();
X509ExtendedTrustManager extendedTrustManager = new InsecureX509TrustManager();
trustManagersParameters.setTrustManager(extendedTrustManager);

SSLContextParameters sslContextParameters = new SSLContextParameters();
sslContextParameters.setTrustManagers(trustManagersParameters);
httpComponent.setSslContextParameters(sslContextParameters);

After this we can use our new insecure http4 component in our Camel routes just as the normal http4 component.
The entire custom http4 component:

package nl.schiphol.api.integration.components;

import org.apache.camel.component.http4.HttpComponent;
import org.apache.camel.util.jsse.SSLContextParameters;
import org.apache.camel.util.jsse.TrustManagersParameters;
import org.apache.http.conn.ssl.AllowAllHostnameVerifier;

import javax.enterprise.inject.Produces;
import javax.inject.Named;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.X509ExtendedTrustManager;
import java.net.Socket;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;

public class CustomHttp4Component {

    @Produces
    @Named("insecurehttps4")
    public HttpComponent createCustomHttp4Component() {
        HttpComponent httpComponent = new HttpComponent();

        httpComponent.setX509HostnameVerifier(AllowAllHostnameVerifier.INSTANCE);


        TrustManagersParameters trustManagersParameters = new TrustManagersParameters();
        X509ExtendedTrustManager extendedTrustManager = new InsecureX509TrustManager();
        trustManagersParameters.setTrustManager(extendedTrustManager);

        SSLContextParameters sslContextParameters = new SSLContextParameters();
        sslContextParameters.setTrustManagers(trustManagersParameters);
        httpComponent.setSslContextParameters(sslContextParameters);

        return httpComponent;
    }

    private static class InsecureX509TrustManager extends X509ExtendedTrustManager {
        @Override
        public void checkClientTrusted(X509Certificate[] x509Certificates, String s, Socket socket) throws CertificateException {

        }

        @Override
        public void checkServerTrusted(X509Certificate[] x509Certificates, String s, Socket socket) throws CertificateException {

        }

        @Override
        public void checkClientTrusted(X509Certificate[] x509Certificates, String s, SSLEngine sslEngine) throws CertificateException {

        }

        @Override
        public void checkServerTrusted(X509Certificate[] x509Certificates, String s, SSLEngine sslEngine) throws CertificateException {

        }

        @Override
        public void checkClientTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {

        }

        @Override
        public void checkServerTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {

        }

        @Override
        public X509Certificate[] getAcceptedIssuers() {
            return new X509Certificate[0];
        }
    }
}

 
2 Comments

Posted by on 2017-02-17 in Geen categorie

 

Tags: , , ,