Code Foundry - Tutorials on Full-Stack Software Development Full-Stack software development

There are many different variants of #batchUpdate methods available in JdbcTemplate. We will specifically look into those who uses BatchPreparedStatementSetter and ParameterizedPreparedStatementSetter.

What is BatchPreparedStatementSetter?

It is an interface used by JdbcTemplate to execute batch updates. This has methods to determine the batch size and method to set parameters in the PreparedStatement. Using this, JdbcTemplate will run only execute single batch based on the batch size returned by implementation this interface.

How to use BatchPreparedStatementSetter?

Let's create a ProductBatchPreparedStatementSetter which can set parameters in the statement.

public class ProductBatchPreparedStatementSetter implements BatchPreparedStatementSetter {

  private final List products;

  public ProductBatchPreparedStatementSetter(List products) {
    Objects.requireNonNull(products);
    
    // Ideally you should do a defensive copy of this list.
    // this.products = new ArrayList<>(products);
    this.products = products;
  }

  @Override
  public void setValues(PreparedStatement ps, int i) throws SQLException {
    Product product = products.get(i);
    ps.setString(1, product.getName());
    ps.setString(2, product.getCategory());
    ps.setString(3, product.getDescription());
  }

  @Override
  public int getBatchSize() {
    return products.size();
  }

}

Usage

int[] results = jdbcTemplate.batchUpdate("insert into product (name, category, description) values(?,?,?)", 
    new ProductBatchPreparedStatementSetter(Arrays.asList(new Product("Lenovo Laptop", "laptop", "Thinkpad series laptop"),
 new Product("Acer Laptop", "laptop", "Predator series laptop"))));
log.info(() -> String.format("Inserted rows: %s", Arrays.toString(results)));

What is ParameterizedPreparedStatementSetter?

It is an interface used by JdbcTemplate to execute batch updates. It has only one method which takes PreparedStatement and Typed object as parameters. Using this, JdbcTemplate can execute multiple batch based on the batch size passed in the #batchUpdate method.

How to use ParameterizedPreparedStatementSetter?

Let's create a pretty straightforward implementation of this interface for our Product example.

ParameterizedPreparedStatementSetter<Product> pss = (ps, product) -> {
    ps.setString(1, product.getName());
    ps.setString(2, product.getCategory());
    ps.setString(3, product.getDescription());
  };

Usage

int batchSize = 5;
int[][] result = jdbcTemplate.batchUpdate("insert into product (name, category, description) values(?,?,?)",
    products, batchSize, pss);
log.info(Arrays.deepToString(result));

#batchUpdate method which uses BatchPreparedStatementSetter returns 1-D int array whereas #batchUpdate method which uses ParameterizedPreparedStatementSetter returns 2-D array. This means that BatchPreparedStatementSetter executed single batch whereas ParameterizedPreparedStatementSetter executed multiple batches.

That's it. You can find the complete code of this example on Github.

What is PreparedStatementSetter?

It is a callback interface used by JdbcTemplate after PreparedStatement is created to set the values in the statement object.

How to use it?

PreparedStatementSetter is also functional interface, so we will use lambda expression in this example to demonstrate PreparedStatementSetter's usage. We will use it in #update method of JdbcTemplate.

int updateCount = jdbcTemplate.update("insert into product(name, category, description) values(?,?,?)", ps -> {
    ps.setString(1, "Lenovo Bag");
    ps.setString(2, "bag");
    ps.setString(3, "Handcrafted bags by Lenovo");
  });
  
  log.info(() -> String.format("Product inserted: %d", updateCount));

You can get the full code of this example from here.

What is ResultSetExtractor?

It is an interface used by #query methods of JdbcTemplate. It is better suitable if you want to map one result object per ResultSet otherwise RowMapper is simpler choice to map one row of ResultSet with one object.

How to use it?

Let's first create a ResultSetExtractor which maps all the rows of ResultSet to single object. For this we will create a ProductResultSetExtractor which returns ProductResponse.

public class ProductResultSetExtractor implements ResultSetExtractor {
  private final RowMapper productRowMapper;
 
  public ProductResultSetExtractor(RowMapper productRowMapper) {
    super();
    this.productRowMapper = productRowMapper;
  }

  @Override
  public ProductResponse extractData(ResultSet rs) throws SQLException {
    final List products = new ArrayList<>();

    int rowNum = 0;
    while(rs.next()) {
      products.add(productRowMapper.mapRow(rs, rowNum));
      rowNum++;
    }

    return ProductResponse.of(products);
  }
}

Now, we will use #query method of JdbcTemplate to use this ProductResultSetExtractor to return result.

ProductResponse productResponse = jdbcTemplate.query("select * from product", new ProductResultSetExtractor(new ProductRowMapper()));

log.info(productResponse::toString);

That's it. You can find the full example code on github.

In this post, we will discuss what RowMapper is and how to use it when writing Jdbc code using Spring JDBC module.

What is RowMapper?

It is an interface of Spring JDBC module which is used by JdbcTemplate to map rows of java.sql.ResultSet. It is typically used when you query data.

Example usage of RowMapper

Let's first create a RowMapper which can map products.

class ProductRowMapper implements RowMapper {

  @Override
    public Product mapRow(ResultSet rs, int rowNum) throws SQLException {
      Product product = new Product();
      product.setId(rs.getInt("id"));
      product.setName(rs.getString("name"));
      product.setDescription(rs.getString("description"));
      product.setCategory(rs.getString("category"));
      return product;
   }
 
}

Now, we will use this ProductRowMapper in #queryForObject of JdbcTemplate.

Product product = jdbcTemplate.queryForObject("select * from product where id=1", new ProductRowMapper());
log.info(product::toString);

You can find the github code here.

What is Spring JdbcTemplate?

JdbcTemplate is the core class of Spring JDBC. It simplifies your interaction with low-level error prone details of JDBC access. You only pass the SQL statement to execute, parameters and processing logic for the returned data and rest is handled by it i.e. Opening Connection, transaction handling, error handling and closing Connection, Statement and Resultset.

How to create object of JdbcTemplate?

1. Calling no args constructor.

JdbcTemplate jdbcTemplate = new JdbcTemplate();

// You need to set datasource in later point in time and also have to call afterPropertiesSet.
jdbcTemplate.setDataSource(DataSource ds);
jdbcTemplate.afterPropertiesSet();

2. By Calling constructor with datasource.

JdbcTemplate jdbcTemplate = new JdbcTemplate(Datasource ds);

3. By Calling constructor with datasource and lazyInit parameter.

JdbcTemplate jdbcTemplate = new JdbcTemplate(DataSource dataSource, boolean lazyInit);

Querying with JdbcTemplate

There are many variants of querying using JdbcTemplate. We will look into queryForObject and queryForList method.

JdbcTemplate.queryForObject(String sql, Class<T> requiredType)

We will use this variant of #queryForObject when the ResultSet returns only single column.

Integer count = jdbcTemplate.queryForObject("select count(*) from product", Integer.class);
log.info(() -> String.format("There are total %d products", count));

JdbcTemplate.queryForObject(String sql, Class requiredType, @Nullable Object... args)

We will use this variant when we need to pass sql binding parameters.

Integer mobileProducts = jdbcTemplate.queryForObject("select count(*) from product where category=?", Integer.class, "mobile");
log.info(() -> String.format("There are total %d mobile products", mobileProducts));

JdbcTemplate.queryForList(String sql, Class elementType)

This variant is useful when query return list of values but for single column.

// E.g. getting list of product names
List mobileNames = jdbcTemplate.queryForList("select name from product where category='mobile'", String.class);
log.info(() -> String.format("Name of mobiles: %s", mobileNames.toString()));

You can get the full example code here.

In the previous post, we have setup ELK stack and ran data analytics on application events and logs. In this post, we will discuss how you can watch real-time application events that are being persisted in the Elasticsearch index and raise alerts if condition for watcher is breached using SentiNL (Kibana plugin).

Few examples of alerting for application events (see previous posts) are:

• Same user logged in from different IP addresses.
• Different users logged in from same IP address.
• PermissionFailures in last 15 minutes.
• Particular kind of exception in last 15 minutes/ hour/ day.

Watching and alerting on Elasticsearch index in Kibana

There are many plugins available for watching and alerting on Elasticsearch index in Kibana e.g. X-Pack, SentiNL.

We will be using SentiNL for watching and alerting on Elasticsearch index.

Installing SentiNL

// Installing SentiNL plugin
/opt/kibana/bin/kibana-plugin --install sentinl -u https://github.com/sirensolutions/sentinl/releases/download/tag-4.6.4-4/sentinl.zip

Configuring SentiNL

SentiNL have wide range of actions that you can configure for watchers. You can send an email, integrate with Slack channel or pushapps, send payload to custom webhook. Open kibana.yml file and add below properties for SentiNL. For our example, we will only enable notification through email.

sentinl:
  es:
    host: 'localhost'
    port: 9200
  settings:
    email:
      active: true
      host: "smtp.gmail.com"
      user: "[EMAIL_ID]"
      password: "[PASSWORD]"
      port: 465
      domain: "gmail.com"
      ssl: true
      tls: false
      authentication: ['PLAIN', 'LOGIN', 'CRAM-MD5', 'XOAUTH2']
      timeout: 20000  # mail server connection timeout
      # cert:
      #   key: '/full/sys/path/to/key/file'
      #   cert: '/full/sys/path/to/cert/file'
      #   ca: '/full/sys/path/to/ca/file'
    slack:
      active: false
      username: 'username'
      hook: 'https://hooks.slack.com/services/'
      channel: '#channel'
    webhook:
      active: false
      host: 'localhost'
      port: 9200
      # use_https: false
      # path: ':/{{payload.watcher_id}}'
      # body: '{{payload.watcher_id}}{payload.hits.total}}'
      # method: POST
    report:
      active: false
      executable_path: '/usr/bin/chromium' # path to Chrome v59+ or Chromium v59+
      timeout: 5000
      # authentication:
      #   enabled: true
      #   mode:
      #     searchguard: false
      #     xpack: false
      #     basic: false
      #     custom: true
      #   custom:
      #     username_input_selector: '#username'
      #     password_input_selector: '#password'
      #     login_btn_selector: '#login-btn'
      # file:
      #   pdf:
      #     format: 'A4'
      #     landscape: true
      #   screenshot:
      #     width: 1280
      #     height: 900
    pushapps:
      active: false
      api_key: ''

That's it!!! Let's start Kibana to configure watcher and alerting in SentiNL.

Creating Watchers and alerting in Kibana

We will be configuring watcher for different users logged in from same IP address and will send e-Mail alerts.

• Open Kibana dashboard on your local machine (Url for Kibana on my local machine is http://localhost:5601).
• Click on SentiNL option in the left nav-pane. You will see a dashboard as below. Click on the New option to create a new watcher.
  
• Click on the Watcher link highlighted as below.
  
• Enter watcher name and schedule in the General tab.
  
• Click on Input tab and enter below mentioned query json in the body. You can also give a name to the query and save.

  {
    "search": {
      "request": {
        "index": [
          "app-events*"
        ],
        "body": {
          "query": {
            "bool": {
              "filter": [
                {
                  "range": {
                    "@timestamp": {
                      "gte": "now-30m"
                    }
                  }
                },
                {
                  "query_string": {
                    "default_field": "appEvent.eventType",
                    "query": "LOGIN_SUCCESS OR LOGIN_FAILURE"
                  }
                }
              ]
            }
          },
          "aggs": {
            "group_by_requestIP": {
              "terms": {
                "field": "appEvent.requestIP.keyword",
                "size": 5
              },
              "aggs": {
                "group_by_identifier": {
                  "terms": {
                    "field": "appEvent.identifier.keyword",
                    "size": 5
                  },
                  "aggs": {
                    "get_latest": {
                      "terms": {
                        "field": "@timestamp",
                        "size": 1,
                        "order": {
                          "_key": "desc"
                        }
                      }
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }
  

• Click on Condition tab and enter below mentioned condition json in the body. You can also give a name to this condition and save.

  {
    "script": {
      "script": "var requestIPbuckets = payload.aggregations.group_by_requestIP.buckets; payload.collector = []; requestIPbuckets.filter(function(requestIP) { return requestIP.key; }).forEach(function(requestIP) { var requestIPKey = requestIP.key; var users = requestIP.group_by_identifier.buckets; if (users.length > 1) { users.filter(function(user) { return user.key; }).forEach(function(user) { payload.collector.push({ 'ip': requestIPKey, 'identifier': user.key, 'count': user.doc_count  }); }); }}); payload.collector.length > 0;"
    }
  }
  

• Click on Action tab and select email as an action for alerting. Give title, to, from, subject and add below mentioned content in the body of email.

  Found {{payload.collector.length}} Events
  {{#payload.collector}}
  {{#.}}
  ip : {{ip}}, identifier: {{identifier}}, count: {{count}}
  {{/.}}
  {{/payload.collector}}
  

• Save the watcher.

This watcher will run periodically based on the schedule that you have set and if the condition for breach is met, will send an email alert. The configured email looks like below.

This is how you can watch real-time changing data in Elasticsearch index and raise alerts based on the configured conditions.

In this post, we will learn how to use Elasticsearch, Logstash and Kibana for running analytics on application events and logs. Firstly, I will install all these applications on my local machine.

Installations

You can read my previous posts on how to install Elasticsearch, Logstash, Kibana and Filebeat on your local machine.

Basic configuration

I hope by now you are have installed Elasticsearch, Logstash, Kibana and Filebeat on your system. Now, Let's do few basic configurations required to be able to run analytics on application events and logs.

Elasticsearch

Open elasticsearch.yml file in [ELASTICSEARCH_INSTLLATION_DIR]/config folder and add properties to it.

cluster.name: gauravbytes-event-analyzer
node.name: node-1

Cluster name is used by Elasticsearch node to form a cluster. Node name within cluster need to be unique. We are running only single instance of Elasticsearch on our local machine. But, in production grade setup there will be master nodes, data nodes and client nodes that you will be configuring as per your requirements.

Logstash

Open logstash.yml file in [LOGSTASH_INSTALLATION_DIR]/config folder and add below properties to it.

node.name: gauravbytes-logstash
path.data: [MOUNTED_HDD_LOCATION]
config.reload.automatic: true
config.reload.interval: 30s

Creating logstash pipeline for parsing application events and logs

There are three parts in pipeline. i.e. input, filter and output. Below the pipeline conf for parsing application event and logs.

input {
    beats {
        port => "5044"
    }
}

filter {
   
    grok {
        match => {"message" => "\[%{TIMESTAMP_ISO8601:loggerTime}\] *%{LOGLEVEL:level} *%{DATA:loggerName} *- (?(.|\r|\n)*)"}
    }
 
    if ([fields][type] == "appevents") {
        json {
            source => "event"
            target => "appEvent"
        }
  
        mutate { 
            remove_field => "event"
        }

        date {
            match => [ "[appEvent][eventTime]" , "ISO8601" ]
            target => "@timestamp"
        }
  
        mutate {
            replace => { "[type]" => "app-events" }
        }
    }
    else if ([fields][type] == "businesslogs") {  
        mutate {
            replace => { "[type]" => "app-logs" }
        }
    }
 
    mutate { 
        remove_field => "message"
    }
}
output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        index => "%{type}-%{+YYYY.MM.dd}"
    }
}

In the input section, we are listening on port 5044 for beat (filebeat to send data on this port).

In the output section, we are persisting data in Elasticsearch on an index based on type and date combination.

Let's discuss the filter section in detail.

• 1) We are using grok filter plugin to parse plain lines of text to structured data.

  grok {
      match => {"message" => "\[%{TIMESTAMP_ISO8601:loggerTime}\] *%{LOGLEVEL:level} *%{DATA:loggerName} *- (?(.|\r|\n)*)"}
  }
  

• 2) We are using json filter plugin to the convert event field to a json object and storing it in appEvent field.

  json {
      source => "event"
      target => "appEvent"
  }
  

• 3) We are using mutate filter plugin to the remove data we don't require.

  mutate { 
      remove_field => "event"
  }
  
  mutate { 
      remove_field => "message"
  }
  

• 4) We are using date filter plugin to the parse the eventTime from appEvent field to ISO8601 dateformat and then replacing its value with @timestamp field..

  date {
      match => [ "[appEvent][eventTime]" , "ISO8601" ]
      target => "@timestamp"
  }
  

Filebeat

Open the file filebeat.yml in [FILEBEAT_INSTALLATION_DIR] and below configurations.

filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - E:\gauravbytes-log-analyzer\logs\AppEvents.log
  fields:
    type: appevents
  
- type: log
  enabled: true
  paths:
    - E:\gauravbytes-log-analyzer\logs\GauravBytesLogs.log
  fields:
    type: businesslogs
  multiline.pattern: ^\[
  multiline.negate: true
  multiline.match: after

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 3

output.logstash:
  hosts: ["localhost:5044"]

In the configurations above, we are defining two different type of filebeat prospectors; one for application events and the other for application logs. We have also defined that the output should be sent to logstash. There are many other configurations that you can do by referencing filebeat.reference.yml file in the filebeat installation directory.

Kibana

Open the kibana.yml in [KIBANA_INSTALLATION_DIR]/config folder and add below configuration to it.

elasticsearch.url: "http://localhost:9200"

We have only configured Elasticsearch url but you can change Kibana host, port, name and other ssl related configurations.

Running ELK stack and Filebeat

//running elasticsearch on windows
\bin\elasticsearch.exe

// running logstash
bin\logstash.bat -f config\gauravbytes-config.conf --config.reload.automatic

//running kibana
bin\kibana.bat

//running filebeat
filebeat.exe -e -c filebeat-test.yml -d "publish"

Creating Application Event and Log structure

I have created two classes AppEvent.java and AppLog.java which will capture information related to application events and logs. Below is the structure for both the classes.

//AppEvent.java
public class AppEvent implements BaseEvent<AppEvent> {
    public enum AppEventType {
        LOGIN_SUCCESS, LOGIN_FAILURE, DATA_READ, DATA_WRITE, ERROR;
    }

    private String identifier;
    private String hostAddress;
    private String requestIP;
    private ZonedDateTime eventTime;
    private AppEventType eventType;
    private String apiName;
    private String message;
    private Throwable throwable;
}

//AppLog.java
public class AppLog implements BaseEvent<AppLog> {
    private String apiName;
    private String message;
    private Throwable throwable;
}

Let's generate events and logs

I have created a sample application to generate dummy events and logs. You can check out the full project on github. There is a AppEventGenerator java file. Run this class with system argument -DLOG_PATH=[YOUR_LOG_DIR] to generate dummy events. If your log_path is not same as one defined in the filebeat-test.yml, then copy the log files generated by this project to the location defined in the filebeat-test.yml. You soon see the events and logs got persisted in the Elasticsearch.

Running analytics on application events and logs in Kibana dashboard

Firstly, we need to define Index pattern in Kibana to view the application events and logs. Follow step by step guide below to create Index pattern.

• Open Kibana dashboard by opening the url (http://localhost:5601/).
  
• Go to Management tab. (Left pane, last option)
  
• Click on Index Patterns link.
  
• You will see already created index, if any. On the left side, you will see Option to Create Index pattern. Click on it.
  
• Now, define index pattern and Click next. Choose time filter field name. I choose @timestamp field for this. You can select any other timestamp field present in this Index and finally click on Create index pattern button.

Let's view Kibana dashboard

Once Index pattern is created, click on Discover tab on the left pane and select index pattern created by you in the previous steps.

You will see a beautiful GUI with a lot of options to mine the data. On the top most pane, you will see option to Auto refresh and data that you would want to fetch (Last 15 minutes, 30 minutes, 1 hour, 1 day and so on) and it will automatically refresh the dashboard.

The next lane has search box. You can further write queries to have more granular view of the data. It uses Apache Lucene's query syntax.

You can also define filters to have a more granular view of data.

This is how you can run the analytics using ELK on your application events and logs. You can also define complex custom filters, queries and create visualization dashboard. Feel free to explore Kibana's official documentation to use it to its full potential.