Microservices With Apache Camel

gowebing

The Apache Camel community introduced a new release a few months ago with a set of components for building microservices. Now, it has support for Hystrix Circuit Breaker and other solutions from Netflix OSS, like the Ribbon load balancer. There is also distributed message tracing with a Zipkin component, as well as service registration and discovery with Consul, etcd, Kubernetes, and Netflix Ribbon.

The new key component for microservices support is ServiceCall EIP, which allows calling a remote service in a distributed system where the service is looked up from a service registry on Kubernetes, Openshift, Cloud Foundry, Zuul, Consul, Zookeeper, etc. In addition, it is worth mentioning that now Apache Camel compiles against Java 8 and has improved support for Spring Boot. A full list of changes is available on Camel community site.

The main purpose of this article is to show how to create microservices with Apache Camel in the most common way using its new features available from release 2.18. In the picture below, you can see the architecture of the proposed solution. The sample application source code is available on GitHub.

This article can be treated as a continuation of my previous blog entry representing a basic example of how to implement microservices in Camel.

Image title

Swagger

Swagger is a powerful open-source framework backed by a large ecosystem of tools that helps you in designing, building, and documenting RESTful APIs. It can be integrated with Camel by adding the dependency camel-swagger or  camel-swagger-java to Maven pom.xml. Then, we can easily configure it with Rest DSL like in the code fragment below.

In our sample, Swagger API is configured on gateway available under the 8000 port. 

restConfiguration()
  .component("netty4-http")
  .bindingMode(RestBindingMode.json)
  .port(8000)
  .apiContextPath("/api-doc")
  .apiProperty("api.title", "Sample Camel API").apiProperty("api.version", "1.0")
  .apiProperty("cors", "true");

With Camel Rest DSL, Swagger configuration is really easy. We only have to call some DSL methods on every operation we would have described.

Finally, we can display the generated documentation by calling http://localhost:8000/api-doc for JSON format and http://localhost:8000/api-doc/swagger.yaml for YAML.

.get("/{id}")
  .description("Find account by id").outType(Account.class)
  .param().name("id").type(RestParamType.path).description("Account identificator").dataType("int").endParam()

Gateway

In our architecture gateway providing API documentation using Swagger framework, we do service call and discovery with the Consul server and ServiceCall EIP pattern. It also has to proxy and load balance requests between all microservice instances. For that functionality, there is no need to search for any external tool — it can be provided using Camel Rest DSL.

You also can use a simple Camel route with Rest Component, but in our sample, we decided to document endpoints using the Swagger framework — which has really smart support in Rest DSL.

Here's route configuration for the API gateway. Load balancing is enabled by default on all routes with the round robin policy. In the code fragment below, you can also see the configuration for connection with Consul, which is running on a Docker container. During route execution, Gateway tries to find all services registered on Consul with the name depending on the route and then selects one for the processing request.

@Component
public class RouteGateway extends RouteBuilder {

@Autowired
CamelContext context;

  @Override
  public void configure() throws Exception {

    ConsulConfigurationDefinition config = new ConsulConfigurationDefinition();
    config.setComponent("netty4-http");
    config.setUrl("http://192.168.99.100:8500");
    context.setServiceCallConfiguration(config);

    restConfiguration()
      .component("netty4-http")
      .bindingMode(RestBindingMode.json)
      .port(8000)
      .apiContextPath("/api-doc")
      .apiProperty("api.title", "Example API").apiProperty("api.version", "1.0")
      .apiProperty("cors", "true");

    JacksonDataFormat formatAcc = new JacksonDataFormat(Account.class);
    JacksonDataFormat formatCus = new JacksonDataFormat(Customer.class);
    JacksonDataFormat formatAccList = new JacksonDataFormat(Account.class);
    formatAccList.useList();
    JacksonDataFormat formatCusList = new JacksonDataFormat(Customer.class);
    formatAccList.useList();

    rest("/account")
    .get("/{id}").description("Find account by id").outType(Account.class)
      .param().name("id").type(RestParamType.path).description("Account identificator").dataType("int").endParam()
      .route().serviceCall("account").unmarshal(formatAcc)
      .endRest()
    .get("/customer/{customerId}").description("Find account by customer id").outType(Account.class)
      .param().name("customerId").type(RestParamType.path).description("Customer identificator").dataType("int").endParam()
      .route().serviceCall("account").unmarshal(formatAcc)
      .endRest()
    .get("/").description("Find all accounts").outTypeList(Account.class)
      .route().serviceCall("account").unmarshal(formatAccList)
      .endRest()
    .post("/").description("Add new account").outTypeList(Account.class)
      .param().name("account").type(RestParamType.body).description("Account JSON object").dataType("Account").endParam()
      .route().serviceCall("account").unmarshal(formatAcc)
      .endRest();

    rest("/customer")
    .get("/{id}").description("Find customer by id").outType(Customer.class)
      .param().name("id").type(RestParamType.path).description("Customer identificator").dataType("int").endParam()
      .route().serviceCall("customer").unmarshal(formatCus)
      .endRest()
    .get("/").description("Find all customers").outTypeList(Customer.class)
      .route().serviceCall("customer").unmarshal(formatCusList)
      .endRest()
    .post("/").description("Add new customer").outTypeList(Customer.class)
      .param().name("customer").type(RestParamType.body).description("Customer JSON object").dataType("Account").endParam()
      .route().serviceCall("customer").unmarshal(formatCus)
      .endRest();

}

}

Configuration with the rest component is much simpler as you see in the code below but without Swagger API.

from("rest:get:account:/{id}").serviceCall("account");
from("rest:get:account:/customer/{customerId}").serviceCall("account");
from("rest:get:account:/").serviceCall("account");
from("rest:post:account:/").serviceCall("account");
from("rest:get:customer:/{id}").serviceCall("customer");
from("rest:get:customer:/").serviceCall("customer");
from("rest:post:customer:/").serviceCall("customer");

Discovery and Registration

Consul is a distributed and highly available system. It is a tool for discovering and configuring services inside infrastructure. It has components for service discovery, health checking, and key/value storage. Consul provides UI management consoles and REST APIs for registering and searching services and key/value objects. REST APIs are available under the v1 path (this is well-documented here).

 I suggest to pulling and running the Docker image with Consul. 

docker run -d --name consul -p 8500:8500 -p 8600:8600 consul

The management console is available under http://192.168.99.100:8500/ui. In the picture below, you can see a list of registered service instances. I'm running two instances of account service and two instances of customer service.

Image title

Service registration with Camel is not as easy, as we could expect. There are no mechanisms out-of-the-box for service registration; there is only a component for a searching service. Fortunately, Consul exposes REST APIs that meet those needs. We are going to use it in our sample application.

Inside the route configuration classes of the account and customer microservices, we configure to the routes that are visible below. 

from("direct:start").routeId("account-consul").marshal().json(JsonLibrary.Jackson)
  .setHeader(Exchange.HTTP_METHOD, constant("PUT"))
  .setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
  .to("http://192.168.99.100:8500/v1/agent/service/register");
from("direct:stop").shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
  .toD("http://192.168.99.100:8500/v1/agent/service/deregister/${header.id}");

They are calling the application and Camel context startup and shutdown. For detecting those phases of the application lifecycle, the EventNotifierSupport  class can be used. Each service is run with the -Dport VM argument, which is injected into the Spring Boot application. The port is used as an identificator during service registration on Consul.

Each service is registered with the ID {service-name}{port}. Here's the EventNotifierSupport implementation from the account service:

@Component
public class EventNotifier extends EventNotifierSupport {

  @Value("${port}")
  private int port;

  @Override
  public void notify(EventObject event) throws Exception {
    if (event instanceof CamelContextStartedEvent) {
      CamelContext context = ((CamelContextStartedEvent) event).getContext();
      ProducerTemplate t = context.createProducerTemplate();
      t.sendBody("direct:start", new Register("account" + port, "account", "127.0.0.1", port));
    }
    if (event instanceof CamelContextStoppingEvent) {
      CamelContext context = ((CamelContextStoppingEvent) event).getContext();
      ProducerTemplate t = context.createProducerTemplate();
      t.sendBodyAndHeader("direct:stop", null, "id", "account" + port);
    }
  }

  @Override
  public boolean isEnabled(EventObject event) {
    return (event instanceof CamelContextStartedEvent || event instanceof CamelContextStoppingEvent);
  }

}

Tracing

The Zipkin web page is a distributed tracing system. It helps gather timing data needed to troubleshoot latency problems in microservice architectures. With the camel-zipkin component, it can be used for tracing and timing incoming and outgoing Camel messages. Events are captured for incoming and outgoing messages being sent via Camel routes.

Integration with it is pretty easy. We need to add a dependency to camel-zipkin-starter in our pom.xml and enable it in the Spring Boot application by adding the annotation @CamelZipkin to our main class.

You also have to configure it in the application.yml file. The full list of component properties is available here. Properties server-service-mappings and client-service-mappings are used for mappings from Camel route to tags displayed in Zipkin UI for each request. You can map it using endpoint or routeId patterns. 

camel:
  springboot:
    name: account-service
    main-run-controller: true
  zipkin:
    host-name: 192.168.99.100
    port: 9410
    server-service-mappings.direct*: consul-acc-reg
    client-service-mappings.direct*: consul-acc-reg
    server-service-mappings.route*: account-service
    client-service-mappings.route*: account-service

In the picture below, there are visible some account-service calls on the Zipkin UI:

Image title

For running Zipkin locally, we are using its Docker image. 

docker run -d --name zipkin -p 9411:9411 -p 9410:9410 openzipkin/zipkin

Microservices

Finally, we are going to take a closer look at some sample microservices: account and customer. Route configuration for account service is rather simple. There are more complications with customer service because we would like to call account service inside the customer GET /customer/{id} method.

To achieve this goal, we need to declare the Consul connection configuration same as on the gateway. Here is the full source code of customer route configuration class. In the GET /customer/{id} method, we would like to find customer data by ID and collect and return in reponse all customer accounts from account service.

In the end of that route, we are calling the enrich DSL method to find and append account to the customer object. This enrichment is realized inside the AggregationStrategyImpl class.

@Component
public class CustomerRoute extends RouteBuilder {

  @Autowired
  CamelContext context;

  @Value("${port}")
  private int port;

  @Override
  public void configure() throws Exception { 

    JacksonDataFormat format = new JacksonDataFormat();
    format.useList();
    format.setUnmarshalType(Account.class);

    ConsulConfigurationDefinition config = new ConsulConfigurationDefinition();
      config.setComponent("netty4-http");
      config.setUrl("http://192.168.99.100:8500");
      context.setServiceCallConfiguration(config);

    restConfiguration()
      .component("netty4-http")
      .bindingMode(RestBindingMode.json)
      .port(port);

    from("direct:start").routeId("account-consul").marshal().json(JsonLibrary.Jackson)
      .setHeader(Exchange.HTTP_METHOD, constant("PUT"))
      .setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
      .to("http://192.168.99.100:8500/v1/agent/service/register");
    from("direct:stop").shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
      .toD("http://192.168.99.100:8500/v1/agent/service/deregister/${header.id}");

    rest("/customer")
      .get("/")
      .to("bean:customerService?method=findAll")
      .post("/").consumes("application/json").type(Customer.class)
      .to("bean:customerService?method=add(${body})")
      .get("/{id}").to("direct:account");

    from("direct:account")
      .to("bean:customerService?method=findById(${header.id})")
      .log("Msg: ${body}").enrich("direct:acc", new AggregationStrategyImpl());

    from("direct:acc").setBody().constant(null).serviceCall("account//account").unmarshal(format);

  }

}

Here's the AggregationStrategyImpl class code:

public class AggregationStrategyImpl implements AggregationStrategy {

  @SuppressWarnings("unchecked")
  public Exchange aggregate(Exchange original, Exchange resource) {
    Object originalBody = original.getIn().getBody();
    Object resourceResponse = resource.getIn().getBody();
    Customer customer = (Customer) originalBody;
    customer.setAccounts((List<Account>) resourceResponse);
    return original;
  }

}

Testing

All of the modules in this sample are implemented with Spring Boot. Before running each of them, you need to start Docker images of Consul and Zipkin. When you are launching account and customer service, you need to set the -Dport VM argument. You can launch more than one instance on different ports. All of them will be registered in Consul. Finally, run Gateway and some endpoints in your web browser or REST client.

Conclusion

I'm positively surprised by Apache Camel. Before I started working on the sample described in this post, I didn’t expect it to have so many features for building microservice solutions. Working with them is simple and fast. Of course, I'm also able to find some disadvantages — like inaccuracies or errors in documentation, only short description of some new components in the developer guide, and no registration process in the discovery server like Consul. In those areas, I clearly see the advantages of the Spring framework. On the other hand, Camel has support for some useful tools like etcd and Kubernetes, which is not available in Spring. In conclusion, I’m looking forward to more improvements of Camel components for building microservices.

 

 

 

 

Top