Add a custom label to the existing metrics in Akka HTTP application

In my previous blog, we discussed to add a custom new metric in Akka HTTP application. If you are looking to expose the promethesu metrics in an Akka HTTP application, please go through to this blog.

Here, in this blog we will discuss to add a custom label in the existing merics. Few days ago, I got a requirement to add a new label to the existing metrics. Let’s discuss that.

The solution was not the straight forward. I had to rewrite a few library classes and then was able to add the custom new label. I was looking to add userId label.

First, we need to rewrite the PrometheusRegistry class and need to follow the below steps

Add userIdLabel to the labels variable:

  private val labels: Seq[String] = {
    val methodLabel = if (settings.includeMethodDimension) Some(MethodDimension.Key) else None
    val pathLabel = if (settings.includePathDimension) Some(PathDimension.Key) else None
    val statusLabel = if (settings.includeStatusDimension) Some(StatusGroupDimension.Key) else None
    val userIdLabel = Some("userId")
    (methodLabel ++ pathLabel ++ statusLabel ++ userIdLabel).toSeq
  }

Add userIdDim to the dimensions:

 response.foreach { r =>
      // compute dimensions
      // format: off
      val methodDim = if (settings.includeMethodDimension) Some(MethodDimension(request.method)) else None
      val pathDim = if (settings.includePathDimension) Some(PathDimension(pathLabel(r))) else None
      val statusGroupDim = if (settings.includeStatusDimension) Some(StatusGroupDimension(r.status)) else None
      val userIdDim = Some(UserIdDimension(userIdLabel(r)))
      val dimensions = (methodDim ++ pathDim ++ statusGroupDim++ userIdDim).toSeq
      
      active.dec()
      responses.inc(dimensions)
      duration.observe(Deadline.now - start, dimensions)
      if (settings.defineError(r)) {
        errors.inc(dimensions)
      }
      r.entity.contentLengthOption.foreach(sentBytes.update(_, dimensions))
    }

Both the above things have been added in a newly written class PrometheusRegistryCustom.

Then, need to define the UserIdLabelHeader class as follows:

final case class UserIdLabelHeader(value: String) extends ModeledCustomHeader[UserIdLabelHeader] {
  override def renderInRequests = false

  override def renderInResponses = false

  override val companion = UserIdLabelHeader
}

object UserIdLabelHeader extends ModeledCustomHeaderCompanion[UserIdLabelHeader] {

  val UnKnown: UserIdLabelHeader = UserIdLabelHeader("UnKnown")

  override val name = "x-userid-label"

  override def parse(value: String): Try[UserIdLabelHeader] = Success(new UserIdLabelHeader(value))
}

Then, need to define the marshaller for PrometheusRegistryCustom class as follows:

trait PrometheusMarshallersCustom {

  val PrometheusContentType: ContentType = {
    MediaTypes.`text/plain` withParams Map("version" -> "0.0.4") withCharset HttpCharsets.`UTF-8`
  }

  implicit val marshaller: ToEntityMarshaller[PrometheusRegistryCustom] = {
    Marshaller.opaque { registry =>
      val output = new StringWriter()
      try {
        TextFormat.write004(output, registry.underlying.metricFamilySamples)
        HttpEntity(output.toString).withContentType(PrometheusContentType)
      } finally {
        output.close()
      }
    }
  }
}

object PrometheusMarshallersCustom extends PrometheusMarshallersCustom

Then, need to define a class (CustomHttpMetricsDirectives) which overrides the HttpMetricsDirectives:

trait CustomHttpMetricsDirectives extends HttpMetricsDirectives {

  def userIdLabel[L](userId: String): Directive[Unit] = {
    extractRequestContext.flatMap { _ =>
      mapResponseHeaders { headers =>
        val userIdHeader = UserIdLabelHeader(userId)
        headers :+ userIdHeader
      }
    }
  }
}

object CustomHttpMetricsDirectives extends CustomHttpMetricsDirectives

At last, there is a class PathLabelHeader which is getting used in PrometheusRegistryCustom class but since this class is private to a particular package so we can not access that. So for that, we have to define a new custom class similar to the PathLabelHeader with different name as PathLabelHeaderCustom:

final case class PathLabelHeaderCustom(value: String) extends ModeledCustomHeader[PathLabelHeaderCustom] {
  override def renderInRequests = false

  override def renderInResponses = false

  override val companion = PathLabelHeaderCustom
}

object PathLabelHeaderCustom extends ModeledCustomHeaderCompanion[PathLabelHeaderCustom] {

  val Unhandled: PathLabelHeaderCustom = PathLabelHeaderCustom("unhandled")
  val UnLabelled: PathLabelHeaderCustom = PathLabelHeaderCustom("unlabelled")

  override val name = "x-path-label"

  override def parse(value: String): Try[PathLabelHeaderCustom] = Success(new PathLabelHeaderCustom(value))
}

Now, use the PrometheusRegistryCustom class in MetricController instead of PrometheusRegistry as follows:

MetricsController.scala

val registry: PrometheusRegistryCustom = PrometheusRegistryCustom(collector, settings)

In the main Api class, use userIdLabel directive as follows:

 def userRoute: Route = get {
    path("user") {
      parameters('userId) { userId =>
        userIdLabel(userId) {
          complete(StatusCodes.OK, "This is an application to add a custom metrics")
        }
      }
    }
  }

Now, run the application and hit the url “http://localhost:8080/user?userId=3” and then check the metrics at this url “http://localhost:8080/admin/prometheus/metrics”

That’s it. I hope this will be really helpful for you guys.

Checkout here to get the full working code.

Add new custom Prometheus metric in an Akka HTTP application

We have already discussed in my previous blog that how to expose the metrics in an Akka http application in prometheus format. In this blog we will discuss, how to add a new custom metric in the Akka http application.

We usually love to customise the things in our own way, similarily we can customise the metrics as per our requirement. Let see how we can achieve that.

Before going into the implementation, let’s understand the requirement first. I need the count of requests per user id. We get the path, status and method as default label in metrics but not the userId. So I decided to write a custom metrics having label as userId which will help me to meet my requirement.

First of all, we need to create a custom counter and then register it with the collector registry.

import com.rishi.metrics.controller.MetricsController.{collector, settings}
import fr.davit.akka.http.metrics.core.Dimension
import io.prometheus.client.Counter

object TotalRequestByUserIdMetric {
  private val userIdLabel: Seq[String] = Seq("userId")

  private val totalRequestCounter: Counter = Counter.build().
    namespace(settings.namespace)
    .name("request_by_user_id")
    .help("Total Request by User ID")
    .labelNames(userIdLabel: _*)
    .register(collector)

  def inc(dimensions: Seq[Dimension]): Unit = {
    totalRequestCounter.labels(dimensions.map(_.value): _*).inc()
  }
}

final case class UserIdDimension(userId: String) extends Dimension {
  override def key: String = "userId"

  override def value: String = userId
}

I have taken “userId” as label name and have added the corrosponding name and help accordingly.

Then, we need to call it from the desired place:

TotalRequestByUserIdMetric.inc(List(UserIdDimension(userId)))

Here, I am calling the inc() function with the userId to increment the counter.

That’s it. We are done. It’s pretty simple.

Let’s hit the below url and see the generated metrics:

http://localhost:8080/user?userId=3

You can get the full code here.

I hope this blog will help you in someway.

Expose Prometheus metrics for an Akka HTTP application

Application monitoring is an essential thing to keep the application available every time. We can do application monitoring through multiple ways such as through healthcheck endpoint, exposing metrics etc. Here, in this blog, we will see, how we can expose the metrics in Prometheus format for an Akka HTTP application.

First, we need to add below maven dependency in pom.xml:

  <dependency>
            <groupId>fr.davit</groupId>
            <artifactId>akka-http-metrics-prometheus_2.12</artifactId>
            <version>1.1.1</version>
    </dependency>

Then, we need to create a MetricController where we need to create a registry and the route for the metrics:

MetricsController:

import akka.http.scaladsl.server.Directives.{path, _}
import akka.http.scaladsl.server.Route
import fr.davit.akka.http.metrics.core.scaladsl.server.HttpMetricsDirectives.metrics
import fr.davit.akka.http.metrics.prometheus.marshalling.PrometheusMarshallers._
import fr.davit.akka.http.metrics.prometheus.{Buckets, PrometheusRegistry, PrometheusSettings, Quantiles}
import io.prometheus.client.CollectorRegistry


class MetricsController {

  import MetricsController._

  val route: Route = (get & path("admin" / "prometheus" / "metrics")) (metrics(registry))
}

object MetricsController {
  val routes: Route = new MetricsController().route

  private val settings: PrometheusSettings = PrometheusSettings
    .default
    .withIncludePathDimension(true)
    .withIncludeMethodDimension(true)
    .withIncludeStatusDimension(true)
    .withDurationConfig(Buckets(1, 2, 3, 5, 8, 13, 21, 34))
    .withReceivedBytesConfig(Quantiles(0.5, 0.75, 0.9, 0.95, 0.99))
    .withSentBytesConfig(PrometheusSettings.DefaultQuantiles)
    .withDefineError(_.status.isFailure)

  private val collector: CollectorRegistry = CollectorRegistry.defaultRegistry

  val registry: PrometheusRegistry = PrometheusRegistry(collector, settings)
}

After this, we need to make the change in the main application file:

Application:

val routes = ApiController.routes ~ MetricsController.routes
val routesWithMetrics = HttpMetricsRoute(routes).recordMetrics(MetricsController.registry)
val bindingFuture = Http().bindAndHandle(routesWithMetrics, "0.0.0.0", 8080)

That’s it. We are done.

Now run the application and hit the below url:

http://localhost:8080/admin/prometheus/metrics

and you will see the metrics as below:

You can see that we are getting metrics as unlabelled for the path. If we have multiple APIs in our service then we would like to label the metrics so that we can easily differentiate the metrics. For that you need to use pathLabeled directive as follows:

pathLabeled("message", "message") {
      complete(StatusCodes.OK, "This is an application to show the way to expose the Prometheus metrics")
    }

First parameter is the label and second parameter is the path.

Now you will get the metrics as follows:

I have used the Prometheus format for the metrics, there are some more formats available in which we can expose the metrics. For more details see here.

I hope, this will be helpful for you guys. Feel free to give your suggestions as the comment to this blog.

References:

https://github.com/RustedBones/akka-http-metrics

Functional Java: Should I really start using the functional paradigm in Java?

I am pretty sure, at some point, you must have had this question in your mind because everywhere, it is functional programming, immutability, higher-order functions, and blah blah.

As a java developer, you must have confused, whether should I move to the functional programming paradigm? What are the benefits it provide to us? People are talking about it everywhere. So let’s give it a try once and then you can decide whether you should go to the functional paradigm or not.

We will see the functional programming features one by one with the code examples and will compare it with the imperative way of java programming.

Immutability:

Mutable variables are poor in taste, and shared mutable variables are pure evil. We often get confused or overlook change to variables. As a result, code with more mutable variables tends to have more errors. Code with shared mutable variables is very hard to parallelize and very hard to reason about. One way to reduce errors is simply to avoid mutability wherever possible, and the functional style makes that easier.

Example: Get the sum of double of each element of a list.

Imperative:

Functional:

In imperative one, we are mutating the sum variable but the functional one is pure immutable.

Declarative:

We have to raise the level of abstraction. Instead of focusing on the “How” part, we should focus on the “What” part. With this, we would be able to focus on the business logic only and can increase productivity.

Example:  Get the sum of double of each element of a list.

Imperative:

Functional:

In imperative one, we are telling to compiler each and every step. How to loop through, from where loop should start, where it should end, double the element and then add it to the sum. A lot of stuff to think about. While in the functional, we just need to think of the business logic. That’s it.

Pure Functions:

A function or method with side effects is hard to understand, hard to maintain, more error-prone, and difficult to parallelize. If we remove side effects, then as long as the input to a function remains unchanged, the output will always be the same.

Having no side effects is critical for referential transparency, which means an invocation or a call to a function can be replaced by its result value without affecting a program’s correctness. With the referential transparency, the compiler can optimize the calls.  Functions that have side effects impose ordering and restrict optimization. On the other hand, calls to functions with no side effects can be moved around and reordered more freely.

Impure Function:

Pure function:

Impure Lambda:

Pure Lambda:

Higher-Order Functions:

In Java 8, one of the biggest changes we have to make is to design with higher-order functions. We’re used to passing objects to methods, return objects from a function but now we have the ability to pass functions as arguments and return a function from a function.

This is because, earlier objects used to be first-class citizens but now functions can also be the first-class citizens. This gives us a more concise code: anywhere we passed anonymous inner classes to a single method interfaces, we can now pass lambda expressions or method references.

Anonymous inner class:

Lambda:

Laziness:

Java already uses lazy execution when evaluating logical operations. For example, in fn1() || fn2() , the call fn2() is never performed until fn1() returns a boolean true. Likewise, if we replace the || with && , the call to fn2() never happens until fn1() returns a boolean false. Programs benefit from this short-circuiting; we avoid unnecessary evaluation of expressions or functions, and that can help improve performance.

Java evaluates logical operators lazily but what if we want method arguments to be evaluated lazily as well. Before Java 8, it was not possible but now with the help of Lambda, we can make it possible. Lambda gets evaluated lazily.

Let’s look into the below class:

We have 2 functions in the above class:

eagerEvaluator() : It evaluates the argument eagerly

lazyEvaluator(): It evaluates the arguments lazily.

Let’s call them one by one and see the output:

eagerEvaluator(evaluate(1), evaluate(2));

Output:

lazyEvaluator(() -> evaluate(1), () -> evaluate(2));

Output:

By looking at the above outputs, we can easily see the difference.

We can get the laziness with the help of Java Streams too. Streams are having 2 types of operations: Intermediate and Terminal. Streams do not get evaluated until we call the terminal operators.

This is all about the functional features in Java.

After looking at the above features, we can easily say that yes, we should start writing code in a functional way in java because we get the following benefits from the code perspective:

• Easy to read and understand
• Easy to maintain
• Easy to reason
• Less error-prone
• Improve testability
• Easy to parallelize

I hope this blog will give you some insights to choose between the imperative and functional programming paradigm in Java.

If you really want to learn more about functional java, see the more code examples here.

References: Functional Programming In java book by Venkat Subramaniam.

Functional Java: How to use a List in a functional way vs the imperative way

In my previous blog, we discussed how we can traverse a list in a functional way. There, first, we saw how we can do it in an imperative way and then moved ahead by modifying the code to reach the functional way. In this blog, we will explore some more uses cases of a list in a functional way and will compare it with the imperative way.

So let’s start.

We will be using the below list here:

final List<String> players = Arrays.asList("Virat", "Rohit",
 "Shikhar", "Rahul", "Rishabh", "Hardik", "MSD");

Transforming a list:

Convert a list of names to all capital letters.

Imperative:

 final List<String> uppercaseNames = new ArrayList();
        for(String player : players) {
            uppercaseNames.add(player.toUpperCase());
        }

Functional:

List<String> uppercaseNames = players.stream()
        .map(String::toUpperCase).collect(Collectors.toList());

In an imperative way, we are creating a new list and then mutating it in the loop while in the functional approach, there is no mutation and the code is very clean.

Filter elements from a list:

From a list of players, let’s pick the ones that start with the letter R.

Imperative:

final List<String> startsWithR = new ArrayList<String>();

for (String player : players) {
    if (player.startsWith("R")) {
        startsWithR.add(player);
    }
}

Functional:

List<String> startsWithR = players.stream().
        filter(player -> player.startsWith("R"))
        .collect(Collectors.toList());

Here, in an imperative way, we are again mutating a newly created list while there is no mutation in a functional way.

Picking an element from the list:

Let’s create a method that will look for an element that starts with a given letter, and prints it

Imperative:

public static void pickName(
        final List<String> players, final String startingLetter) {
    String foundName = null;
    for (String player : players) {
        if (player.startsWith(startingLetter)) {
            foundName = player;
            break;
        }
    }

    System.out.print(String.format("A name starting with %s: ", startingLetter));
    if (foundName != null) {
        System.out.println(foundName);
    } else {
        System.out.println("No name found");
    }
}

Functional:

public static void pickName(
        final List<String> names, final String startingLetter) {
    final Optional<String> foundName =
            names.stream()
                    .filter(name ->name.startsWith(startingLetter))
                    .findFirst();
    System.out.println(String.format("A name starting with %s: %s",
            startingLetter, foundName.orElse("No name found")));
}

You can see, the functional way is very clear and concise. We have used break in an imperative way and then later added the If condition to check the value of foundName variable while in a functional way, it is a very clean and simple approach where we have used Optional.

Joining Elements:

Let’s make a comma-separated string by joining the elements of List and print it.

Imperative:

String names = "";
for (int i = 0; i < players.size() - 1; i++) {
    names += players.get(i) + ", ";
}

if(players.size() > 0)
    names += players.get(players.size() - 1);

System.out.println(names);

Functional:

Approach 1:

String names = players.stream().collect(Collectors.joining(", "));
System.out.println(names);

Approach 2:

String names = String.join(", ", players);
System.out.println(names);

In an imperative approach, we had to use the normal for loop just to iterate before the last element and then later added the last element explicitly just to avoid the extra comma at the end while the functional approach is very neat and clean.

Conclusion:

As a conclusion, we get the following benefits by using functional approach:

• Need to write less code
• Code is maintainable
• No mutability
• More focus on business logic instead of writing the basic coding
• and many more.

That’s it. Hope this blog will be helpful to give you a better understanding of using List in a functinal way. Stay tuned !!!

Functional Java: Traversing a list in a functional way

In this blog, we will see how we can traverse a list in Java in a functional way. Iterating through a list is a basic operation on a collection, but over the years it’s gone through a few significant changes. We’ll begin with the old style and evolve an example—enumerating a list of names—to the elegant style.

Let’s create a list first:

final List<String> players = Arrays.asList(“Virat”, “Rohit”, “Shikhar”, “Rahul”, “Rishabh”, “Hardik”, “MSD”);

Let’s talk about the habitual way first i.e. imperative way.

for(int i = 0; i < players.size(); i++) {
System.out.println(players.get(i));
}

This is the verbose and error-prone way and I personally don’t like it. Here, we need to define how to part as well which we do not need to do in a functional way.

Java also offers a construct that is a bit more civilized than the good old for loop.

for(String name : players) {
System.out.println(name);
}

Under the hood, this form of iteration uses the Iterator interface and calls into its hasNext() and next() methods.

Both these versions are external iterators, which mix how we do it with what we’d like to achieve. We explicitly control the iteration with them, indicating where to start and here to end; the second version does that under the hood using the Iterator methods. With explicit control, the break and continue statements can also help manage the iteration’s flow of control.

The second construct has less ceremony than the first. Its style is better than the first if we don’t intend to modify the collection at a particular index. Both of these styles, however, are imperative and we can dispense with them in modern Java.

There are quite a few reasons to favor the change to the functional style:

• The for loops are inherently sequential and are quite difficult to parallelize.
• Such loops are non-polymorphic; we get exactly what we ask for. We passed the collection to for instead of invoking a method (a polymorphic operation) on the collection to perform the task.
• At the design level, the code fails the “Tell, don’t ask” principle. We ask for a specific iteration to be performed instead of leaving the details of the iteration to underlying libraries.

The Iterable interface has been enhanced in JDK 8 with a special method named forEach(), which accepts a parameter of type Consumer. As the name indicates, an instance of Consumer will consume, through its accept() method, what’s given to it. Let’s use the forEach() method with the all-too-familiar anonymous inner class syntax.

players.forEach(new Consumer<String>() {
public void accept(final String name) {
System.out.println(name);
}
});

We changed just one thing: we traded in the old for loop for the new internal iterator forEach(). As for the benefit, we went from specifying how to iterate to focusing on what we want to do for each element. The bad news is the code looks a lot more verbose.

Let’s make one more change, replacing the anonymous inner class with a lambda expression.

players.forEach((final String name) -> System.out.println(name));

Here, forEach() is a high order function.

Java compiler gives us the flexibility to leave off the type parameter, it will infer by its own.

players.forEach((name) -> System.out.println(name));

The Java compiler treats single-parameter lambda expressions as special. We can leave off the parentheses around the parameter if the parameter’s type is inferred.

players.forEach(name -> System.out.println(name));

There is one more way which we can use with forEach() and that is method reference:

players.forEach(System.out::println);

That’s all about the traversing a list in a functional way. Let’s give it a try once and I am sure you will gonna love this functional way.

There are many other functions with a list which we will be discussing in my upcoming blogs. Stay Tuned.

Reactive Spring: Define a REST endpoint as a continuous stream

In the REST APIs, all Http requests are stateless. We fire the request and get the response, That’s it. It does not keep any state for any HTTP request. The connection between client and server is lost once the transaction ends, so 1 response for 1 request.

But sometimes, we get the requirement to have a continuous response for a single request. This continuous response is called Streaming response. So in this blog, we will see how we can do that in Spring.

Here, I am assuming, we are already having a logic which is creating continuous output. I am just going to show how we can wrap the continuous output in a streaming response and can send it to the client. So let’s begin.

For the continuous output, we will be using the below piece of code:

Flux.interval(Duration.ofSeconds(1));

The above code will produce a sequential long value starting from 0, in the interval of 1 second infinitely.

To send the streaming response, we need to set the produced media type as follows:

MediaType.APPLICATION_STREAM_JSON_VALUE

The full code for GetMapping:

@GetMapping(value = “/streaming”, produces = MediaType.APPLICATION_STREAM_JSON_VALUE)

The full code for the controller:

@RestController
public class StreamingController {

@GetMapping(value = “/streaming”, produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<Long> getItemsStream(){

return Flux.interval(Duration.ofSeconds(1)); // You can write your own logic here.
}
}

That’s it for the controller. It’s very simple.

Now, let’s write the test cases to test the controller behavior.

First, we need to add below dependencies in pom.xml:

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>

<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
</dependency>

Now, write the test file:

@RunWith(SpringRunner.class)
@SpringBootTest
@AutoConfigureWebTestClient
public class StreamingControllerTest {

@Autowired
WebTestClient webTestClient;

@Test
public void fluxStream() {

Flux<Long> longStreamFlux = webTestClient.get().uri(“/streaming”)
.accept(MediaType.APPLICATION_STREAM_JSON)
.exchange()
.expectStatus().isOk()
.returnResult(Long.class)
.getResponseBody();
StepVerifier.create(longStreamFlux)
.expectNext(0l)
.expectNext(1l)
.expectNext(2l)
.thenCancel()
.verify();
}
}

Here, we are using WebTestClient to test the REST endpoint and StepVerifier to test the Flux output. You can see, we are expecting 3 elements and then we have called cancel. As it is a streaming response, so we would not get onComplete event here so we would need to cancel the request explicitly to verify the behavior. You can get the full code here.

I hope, this blog will help you to define the streaming endpoint for Spring.

Access multiple couchbase buckets from a Reactive Spring Boot application

A few days ago, I got a situation where I needed to access more than one couchbase bucket from a single reactive spring boot (Spring Web-flux) application.

It is all about configuration. So, first of all, we will see how we can access a single couchbase bucket and then will move forward to access multiple buckets.

Access Single bucket:

To access a single bucket,  we need to define a couchbase configuration file and need to define some properties in the application.properties file.

CouchBaseConfiguration:

@Configuration
public class CouchBaseConfiguration extends AbstractReactiveCouchbaseConfiguration {

    private final String bucketUsername;
    private CouchbaseProperties couchbaseProperties;

    public CouchBaseConfiguration(CouchbaseProperties couchbaseProperties,
                                  @Value("${spring.couchbase.bucket.username}") String bucketUsername) {
        this.couchbaseProperties = couchbaseProperties;
        this.bucketUsername = bucketUsername;
    }

    @Override
    protected List<String> getBootstrapHosts() {
        return couchbaseProperties.getBootstrapHosts();
    }

    @Override
    protected String getBucketName() {
        return couchbaseProperties.getBucket().getName();
    }

    @Override
    protected String getBucketPassword() {
        return couchbaseProperties.getBucket().getPassword();
    }

    @Override
    protected String getUsername() {
        return bucketUsername;
    }
}

application.properties:

spring.couchbase.bootstrap-hosts=${COUCHBASE_HOST:localhost:8091}
spring.couchbase.bucket.name=${COUCHBASE_BUCKET:club}
spring.couchbase.bucket.username=${COUCHBASE_USERNAME:admin}
spring.couchbase.bucket.password=${COUCHBASE_PASSWORD:123456}

Access multiple buckets:

To access multiple buckets, we need to override configureReactiveRepositoryOperationsMapping() function of AbstractReactiveCouchbaseConfiguration class.

CouchBaseConfiguration:

@Configuration
@Slf4j
public class CouchBaseConfiguration extends AbstractReactiveCouchbaseConfiguration {

    private CouchbaseProperties couchbaseProperties;
    private final String username;
    private final String clubBucketName;
    private final String clubAuthBucketName;

    public CouchBaseConfiguration(CouchbaseProperties couchbaseProperties,
                                  @Value("${spring.couchbase.bucket.username}") String bucketUsername,
                                  @Value("${couchbase.club.bucket.name}") String clubBucketName,
                                  @Value("${couchbase.club.auth.bucket.name}") String clubAuthBucketName) {
        this.couchbaseProperties = couchbaseProperties;
        this.username = bucketUsername;
        this.clubBucketName = clubBucketName;
        this.clubAuthBucketName = clubAuthBucketName;
    }

    @Override
    protected List<String> getBootstrapHosts() {
        return couchbaseProperties.getBootstrapHosts();
    }

    @Override
    protected String getBucketName() {
        return clubBucketName;
    }

    @Override
    protected String getBucketPassword() {
        return couchbaseProperties.getBucket().getPassword();
    }

    @Override
    protected String getUsername() {
        return username;
    }

    private Bucket clubAuthBucket() throws Exception {
        return couchbaseCluster().openBucket(clubAuthBucketName);
    }

    private RxJavaCouchbaseTemplate crewAuthTemplate() throws Exception {
        RxJavaCouchbaseTemplate template = new RxJavaCouchbaseTemplate(
                couchbaseClusterInfo(), clubAuthBucket(),
                mappingCouchbaseConverter(), translationService());
        template.setDefaultConsistency(getDefaultConsistency());
        return template;
    }

    @Override
    public void configureReactiveRepositoryOperationsMapping(ReactiveRepositoryOperationsMapping baseMapping) {
        try {
            baseMapping.mapEntity(CrewChatProfile.class, crewAuthTemplate());
        } catch (Exception ex) {
            log.error("Error in creating mapping for {} bucket", clubAuthBucketName, ex);
        }
    }

}

application.properties:

#couchbase config
spring.couchbase.bootstrap-hosts=${COUCHBASE_BOOTSTRAP_HOSTS:localhost:8091}
spring.couchbase.bucket.username=${COUCHBASE_USERNAME:admin}
spring.couchbase.bucket.password=${COUCHBASE_PASSWORD:123456}
spring.couchbase.env.timeouts.connect=${COUCHBASE_CONNECTION_TIMEOUT:30000}
couchbase.club.bucket.name=${COUCHBASE_CLUB_BUCKET:club}
couchbase.club.auth.bucket.name=${COUCHBASE_CLUB_AUTH_BUCKET:club_auth}

That’s it. I have kept this blog short and just provided only the required information.

I hope, it will be helpful for you.

Functional Java: Let’s understand the higher-order function

Higher-order function is an essential part of the functional programming paradigm. We must have defined a lot of functions in any language where we pass either primitive types or an object as an argument and returns the same. These are normal functions or methods.

Then comes first-class functions. First-class functions are the functions that are treated as values. It means, those functions can be assigned to a variable and can be passed around as an argument.

And finally, comes higher-order functions. These are the functions that either takes one or more function as an argument or returns the function as a result.

The two things (First class function and higher-order function) are closely related, as it’s hard to imagine a language with first-class functions that would not also support higher-order functions, and conversely a language with higher-order functions but without first-class function support.

Before Java8, we used to pass a function to a function with the help of anonymous inner classes, but now we do that with the help of Lambdas.

Before Java 8:

After Java 8:

In the above example,

we are passing pathname -> pathname.getAbsolute().endsWith(“txt”) as an argument to listFiles().

Generally, a function has a body, a name, a parameter list, and a return type. The passed function here has a parameter list followed by an arrow ( -> ), and then the short body. The type of the parameter may be inferred by the Java compiler here and the return type is implicit. This function is anonymous; it has no name.

Rather than referring to these as anonymous functions, we call them lambda expressions.

Let’s understand the higher-order function in deep with some code examples:

Returns function as a result:

In the above example, checkIfStartsWith() returns Predicate.

In the above example, checkIfStartsWith() returns Function which takes 1 argument and returns the Predicate.

In the above example, getList() returns BiFunction which takes 2 arguments and returns the List.

Predicate, Function, and BiFunction are the functional interface that is used to write Lamda expression in the above examples.

Take the function as an argument:

In the above example, filter() is the higher-order function which takes a Predicate as an argument.

In the above example, filter() is the higher-order function that takes Function as an argument.

In the above example, getList() is the higher-order function that takes Predicate as an argument.

We have taken the examples only for 3 different Functional Interfaces, but there are more predefined functional interfaces such as Supplier, Consumer, etc and even we can define our own custom Functional interface and can use that to define Lambda expression.

So that is all about higher-order function. I hope this blog will help you to understand the higher-order function in Java which is enabled in Java 8 with the help of Lamdas and Functional interfaces.

Expose Prometheus metrics for an Akka HTTP application

Application monitoring is an essential thing to keep the application available every time. We can do application monitoring through multiple ways such as through healthcheck endpoint, exposing metrics etc. Here, in this blog, we will see, how we can expose the metrics in Prometheus format for an Akka HTTP application.

First, we need to add below maven dependency in pom.xml:

pom.xml:

    <dependency>
            <groupId>fr.davit</groupId>
            <artifactId>akka-http-metrics-prometheus_2.12</artifactId>
            <version>1.1.1</version>
    </dependency>

Then, we need to create a MetricController where we need to create a registry and the route for the metrics:

MetricsController:

import akka.http.scaladsl.server.Directives.{path, _}
import akka.http.scaladsl.server.Route
import fr.davit.akka.http.metrics.core.scaladsl.server.HttpMetricsDirectives.metrics
import fr.davit.akka.http.metrics.prometheus.marshalling.PrometheusMarshallers._
import fr.davit.akka.http.metrics.prometheus.{Buckets, PrometheusRegistry, PrometheusSettings, Quantiles}
import io.prometheus.client.CollectorRegistry


class MetricsController {

  import MetricsController._

  val route: Route = (get & path("admin" / "prometheus" / "metrics")) (metrics(registry))
}

object MetricsController {
  val routes: Route = new MetricsController().route

  private val settings: PrometheusSettings = PrometheusSettings
    .default
    .withIncludePathDimension(true)
    .withIncludeMethodDimension(true)
    .withIncludeStatusDimension(true)
    .withDurationConfig(Buckets(1, 2, 3, 5, 8, 13, 21, 34))
    .withReceivedBytesConfig(Quantiles(0.5, 0.75, 0.9, 0.95, 0.99))
    .withSentBytesConfig(PrometheusSettings.DefaultQuantiles)
    .withDefineError(_.status.isFailure)

  private val collector: CollectorRegistry = CollectorRegistry.defaultRegistry

  val registry: PrometheusRegistry = PrometheusRegistry(collector, settings)
}

After this, we need to make the change in the main application file:

Application:

val routes = ApiController.routes ~ MetricsController.routes
val routesWithMetrics = HttpMetricsRoute(routes).recordMetrics(MetricsController.registry)
val bindingFuture = Http().bindAndHandle(routesWithMetrics, "0.0.0.0", 9000)

That’s it. We are done.

Now run the application and hit the below url:

http://localhost:9000/admin/prometheus/metrics

and you will see the metrics as below:

You can see that we are getting metrics as unlabelled for the path. If we have multiple APIs in our service then we would like to label the metrics so that we can easily differentiate the metrics. For that you need to use pathLabeled directive as follows:

pathLabeled("message", "message") {
      complete(StatusCodes.OK, "This is an application to show the way to expose the Prometheus metrics")
    }

First parameter is the label and second parameter is the path.

Now you will get the metrics as follows:

I have used the Prometheus format for the metrics, there are some more formats available in which we can expose the metrics. For more details see here.

I hope, this will be helpful for you guys. Feel free to give your suggestions as the comment to this blog.

References:

https://github.com/RustedBones/akka-http-metrics