Skip to content
This repository was archived by the owner on Aug 20, 2025. It is now read-only.

METRON-633: Create better logging for HbaseEnrichmentWriter #572

Closed
zezutom wants to merge 1 commit into from
Closed

METRON-633: Create better logging for HbaseEnrichmentWriter #572

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 37 additions & 6 deletions ...chment/src/main/java/org/apache/metron/enrichment/writer/SimpleHbaseEnrichmentWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,13 +38,18 @@
import org.apache.metron.writer.AbstractWriter;
import org.apache.metron.common.writer.BulkWriterResponse;
import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

public class SimpleHbaseEnrichmentWriter extends AbstractWriter implements BulkMessageWriter<JSONObject>, Serializable {

private static final Logger LOG = LoggerFactory.getLogger(SimpleHbaseEnrichmentWriter.class);

public enum Configurations {
HBASE_TABLE("shew.table")
,HBASE_CF("shew.cf")
Expand All @@ -62,13 +67,18 @@ public String getKey() {
return key;
}
public Object get(Map<String, Object> config) {
return config.get(key);
Object o = config.get(key);
if (o == null) {
LOG.warn("No config object found for key: '{}'", key);
}
return o;
}
public <T> T getAndConvert(Map<String, Object> config, Class<T> clazz) {
Object o = get(config);
if(o != null) {
return ConversionUtils.convert(o, clazz);
}
LOG.warn("No object of type '{}' found in config", clazz);
return null;
}
}
Expand All @@ -94,11 +104,12 @@ public KeyTransformer(Iterable<String> keys, String delim) {
}

public String transform(final JSONObject message) {
return
keys.stream().map( x -> {
String transformedMessage = keys.stream().map(x -> {
Object o = message.get(x);
return o == null?"":o.toString();
}).collect(Collectors.joining(delim));
LOG.debug("Transformed message: '{}'", transformedMessage);
return transformedMessage;
}
}
private transient EnrichmentConverter converter;
Expand All @@ -120,6 +131,11 @@ public void configure(String sensorName, WriterConfiguration configuration) {
if(converter == null) {
converter = new EnrichmentConverter();
}
LOG.debug("Sensor: '{}': {Provider: '{}', Converter: '{}'}", sensorName, getClassName(provider), getClassName(converter));
}

private String getClassName(Object object) {
return object == null ? "" : object.getClass().getName();
}

@Override
Expand Down Expand Up @@ -150,6 +166,7 @@ public HTableInterface getTable(String tableName, String cf) throws IOException
if(table != null) {
table.close();
}
LOG.debug("Fetching table '{}', column family: '{}'", tableName, cf);
table = getProvider().getTable(conf, tableName);
this.tableName = tableName;
this.cf = cf;
Expand All @@ -169,16 +186,19 @@ public HTableInterface getTable(Map<String, Object> config) throws IOException {
private List<String> getColumns(Object keyColumnsObj, boolean allowNull) {
Object o = keyColumnsObj;
if(allowNull && keyColumnsObj == null) {
LOG.debug("No key columns were specified");
return Collections.emptyList();
}
if(o instanceof String) {
LOG.debug("Key column: '{}'", o);
return ImmutableList.of(o.toString());
}
else if (o instanceof List) {
List<String> keyCols = new ArrayList<>();
for(Object key : (List)o) {
keyCols.add(key.toString());
}
LOG.debug("Key columns: '{}'", String.join(",", keyCols));
return keyCols;
}
else {
Expand All @@ -190,14 +210,17 @@ private KeyTransformer getTransformer(Map<String, Object> config) {
Object o = Configurations.KEY_COLUMNS.get(config);
KeyTransformer transformer = null;
if(keyTransformer != null && keyTransformer.getKey() == o) {
return keyTransformer.getValue();
transformer = keyTransformer.getValue();
LOG.debug("Transformer found for key '{}': '{}'", o, transformer);
return transformer;
}
else {
List<String> keys = getColumns(o, false);
Object delimObj = Configurations.KEY_DELIM.get(config);
String delim = (delimObj == null || !(delimObj instanceof String))?null:delimObj.toString();
transformer = new KeyTransformer(keys, delim);
keyTransformer = new AbstractMap.SimpleEntry<>(o, transformer);
LOG.debug("Transformer found for keys '{}' and delimiter '{}': '{}'", String.join(",", keys), delim, transformer);
return transformer;
}
}
Expand All @@ -213,7 +236,7 @@ private EnrichmentValue getValue( JSONObject message
for (Object kv : message.entrySet()) {
Map.Entry<Object, Object> entry = (Map.Entry<Object, Object>) kv;
if (!keyColumns.contains(entry.getKey())) {
metadata.put(entry.getKey().toString(), entry.getValue());
addMetadataEntry(metadata, entry);
}
}
return new EnrichmentValue(metadata);
Expand All @@ -222,13 +245,20 @@ private EnrichmentValue getValue( JSONObject message
for (Object kv : message.entrySet()) {
Map.Entry<Object, Object> entry = (Map.Entry<Object, Object>) kv;
if (valueColumns.contains(entry.getKey())) {
metadata.put(entry.getKey().toString(), entry.getValue());
addMetadataEntry(metadata, entry);
}
}
return new EnrichmentValue(metadata);
}
}

private void addMetadataEntry(Map<String, Object> metadata, Map.Entry<Object, Object> entry) {
String key = entry.getKey().toString();
Object value = entry.getValue();
LOG.debug("Adding metadata: {Key: '{}', Value: '{}'}", key, value);
metadata.put(key, value);
}

private EnrichmentKey getKey(JSONObject message, KeyTransformer transformer, String enrichmentType) {
if(enrichmentType != null) {
return new EnrichmentKey(enrichmentType, transformer.transform(message));
Expand Down Expand Up @@ -260,6 +290,7 @@ public BulkWriterResponse write(String sensorType
}
Put put = converter.toPut(this.cf, key, value);
if(put != null) {
LOG.debug("Put: {Column Family: '{}', Key: '{}', Value: '{}'}", this.cf, key, value);
puts.add(put);
}
}
Expand Down
176 changes: 176 additions & 0 deletions ...rsers/src/test/java/org/apache/metron/writers/SimpleHBaseEnrichmentWriterLoggingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,176 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.metron.writers;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.metron.common.configuration.writer.WriterConfiguration;
import org.apache.metron.enrichment.converter.EnrichmentConverter;
import org.apache.metron.enrichment.integration.mock.MockTableProvider;
import org.apache.metron.enrichment.writer.SimpleHbaseEnrichmentWriter;
import org.json.simple.JSONObject;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

import static org.junit.Assert.*;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.verify;
import static org.powermock.api.mockito.PowerMockito.*;

@RunWith(PowerMockRunner.class)
@PrepareForTest({SimpleHbaseEnrichmentWriter.class, LoggerFactory.class, HBaseConfiguration.class})
public class SimpleHBaseEnrichmentWriterLoggingTest {

private static Logger logger;

private static final String COLUMN_FAMILY = "cf1";

private SimpleHbaseEnrichmentWriter hbaseEnrichmentWriter;

@BeforeClass
public static void init() {
mockStatic(LoggerFactory.class);
logger = mock(Logger.class);
when(LoggerFactory.getLogger(SimpleHbaseEnrichmentWriter.class)).thenReturn(logger);
}

@Before
public void setUp() {
hbaseEnrichmentWriter = new SimpleHbaseEnrichmentWriter();
mockStatic(HBaseConfiguration.class);
Configuration mockConfig = mock(Configuration.class);
when(mockConfig.getInt(eq("hbase.column.max.version"), anyInt())).thenReturn(1);
when(HBaseConfiguration.create()).thenReturn(mockConfig);
}

@Test
public void shouldWarnOnMissingConfig() {
SimpleHbaseEnrichmentWriter.Configurations config = getHbaseTableConfig();
config.get(Collections.emptyMap());
verify(logger).warn("No config object found for key: '{}'", config.getKey());
}

@Test
public void shouldWarnOnMissedConfigConversion() {
getHbaseTableConfig().getAndConvert(Collections.emptyMap(), String.class);
verify(logger).warn("No object of type '{}' found in config", String.class);
}

@Test
public void shouldDebugTransformedMessage() {

// An expected property
String key = "message";
String value = "Hello World!";

// A received message bearing the property
SimpleHbaseEnrichmentWriter.KeyTransformer keyTransformer = new SimpleHbaseEnrichmentWriter.KeyTransformer(key);
JSONObject message = newJSONObject(key, value);

// Message transformation
keyTransformer.transform(message);

// Proof the result has been captured
verify(logger).debug("Transformed message: '{}'", value);
}

@Test
public void shouldDebugSensorConfig() {
String sensorName = "someSensor";
hbaseEnrichmentWriter.configure(sensorName, getMockWriterConfiguration());
verify(logger).debug("Sensor: '{}': {Provider: '{}', Converter: '{}'}",
sensorName, MockTableProvider.class.getName(), EnrichmentConverter.class.getName());
}

@Test
public void shouldDebugFetchedHBaseTable() throws IOException {
String sensorName = "someSensor";
String tableName = "someTable";
hbaseEnrichmentWriter.configure(sensorName, getMockWriterConfiguration());
hbaseEnrichmentWriter.getTable(tableName, COLUMN_FAMILY);
verify(logger).debug("Fetching table '{}', column family: '{}'", tableName, COLUMN_FAMILY);
}

@Test
public void shouldDebugWriteOperation() throws Exception {
String sensorName = "someSensor";
String key = "testKey";
String value = "testValue";
WriterConfiguration mockWriterConfiguration = getMockWriterConfiguration();
hbaseEnrichmentWriter.configure(sensorName, mockWriterConfiguration);
hbaseEnrichmentWriter.write(sensorName,
mockWriterConfiguration,
Collections.emptyList(),
Collections.singletonList(newJSONObject(key, value)));

ArgumentCaptor<String> logTemplateCaptor = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<Object> logArgsCaptor = ArgumentCaptor.forClass(Object.class);
verify(logger, atLeastOnce()).debug(logTemplateCaptor.capture(),
logArgsCaptor.capture(), logArgsCaptor.capture(), logArgsCaptor.capture());

assertEquals("Put: {Column Family: '{}', Key: '{}', Value: '{}'}", logTemplateCaptor.getValue());

List<Object> logArgs = logArgsCaptor.getAllValues();
assertEquals(COLUMN_FAMILY, assertAndGetArg(logArgs, 3));
assertEquals("EnrichmentKey{indicator=':', type='testEnrichment'}", assertAndGetArg(logArgs, 2).toString());
assertEquals("EnrichmentValue{metadata={testKey=testValue}}", assertAndGetArg(logArgs, 1).toString());
}

private <T> T assertAndGetArg(List<Object> capturedArgs, int index) {
assertTrue(capturedArgs.size() - index >= 0);
T result = (T) capturedArgs.get(capturedArgs.size() - index);
assertNotNull(result);
return result;
}

private JSONObject newJSONObject(String key, String value) {
return new JSONObject(Collections.singletonMap(key, value));
}

private SimpleHbaseEnrichmentWriter.Configurations getHbaseTableConfig() {
return SimpleHbaseEnrichmentWriter.Configurations.HBASE_TABLE;
}

private WriterConfiguration getMockWriterConfiguration() {
WriterConfiguration writerConfiguration = mock(WriterConfiguration.class);
Map<String, Object> configMap = new HashMap<>();

String tableName = SimpleHBaseEnrichmentWriterLoggingTest.class.getSimpleName();
MockTableProvider.addTable(tableName, COLUMN_FAMILY);

configMap.put(SimpleHbaseEnrichmentWriter.Configurations.HBASE_PROVIDER.getKey(), MockTableProvider.class);
configMap.put(SimpleHbaseEnrichmentWriter.Configurations.KEY_COLUMNS.getKey(), Arrays.asList("col1", "col2"));
configMap.put(SimpleHbaseEnrichmentWriter.Configurations.HBASE_TABLE.getKey(), tableName);
configMap.put(SimpleHbaseEnrichmentWriter.Configurations.HBASE_CF.getKey(), COLUMN_FAMILY);
configMap.put(SimpleHbaseEnrichmentWriter.Configurations.ENRICHMENT_TYPE.getKey(), "testEnrichment");
when(writerConfiguration.getSensorConfig(anyString())).thenReturn(configMap);
return writerConfiguration;
}
}