move TcpIngestor to pdb-ui

and start it in the web application.
Also use the spring way of handling property files.

pdb-ui/src/main/java/org/lucares/pdbui/Ingestion.java

@@ -0,0 +1,21 @@
+package org.lucares.pdbui;
+
+import javax.annotation.PostConstruct;
+
+import org.springframework.stereotype.Component;
+
+@Component
+public class Ingestion {
+
+	private final Ingestor tcpIngestor;
+
+	public Ingestion(final Ingestor tcpIngestor) {
+		this.tcpIngestor = tcpIngestor;
+	}
+
+	@PostConstruct
+	public void start() throws Exception {
+		tcpIngestor.start();
+	}
+
+}

pdb-ui/src/main/java/org/lucares/pdbui/Ingestor.java

@@ -0,0 +1,7 @@
+package org.lucares.pdbui;
+
+public interface Ingestor {
+
+	void start() throws Exception;
+
+}

pdb-ui/src/main/java/org/lucares/pdbui/MySpringConfiguration.java

@@ -2,12 +2,11 @@ package org.lucares.pdbui;
 
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
+import org.springframework.scheduling.annotation.EnableAsync;
 
 @Configuration
+@EnableAsync
 @ComponentScan("org.lucares.pdbui")
-@PropertySource("classpath:/config.system.properties")
-@PropertySource("classpath:/config.user.properties")
 public class MySpringConfiguration {
 
 }

pdb-ui/src/main/java/org/lucares/pdbui/PdbRepository.java

@@ -6,16 +6,23 @@ import java.util.List;
 
 import org.lucares.ludb.Proposal;
 import org.lucares.performance.db.PerformanceDb;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Repository;
 
 @Repository
 public class PdbRepository implements DisposableBean {
+	private static final Logger LOGGER = LoggerFactory.getLogger(PdbRepository.class);
+
 	private final PerformanceDb db;
 
 	public PdbRepository(@Value("${db.base}") final String dbBaseDir) {
 		final Path dataDirectory = Paths.get(dbBaseDir);
+
+		LOGGER.info("using database in {}", dataDirectory.toAbsolutePath());
+
 		this.db = new PerformanceDb(dataDirectory);
 	}
 

pdb-ui/src/main/java/org/lucares/pdbui/TcpIngestor.java

@@ -0,0 +1,278 @@
+package org.lucares.pdbui;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.PrintWriter;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.SocketTimeoutException;
+import java.nio.file.Path;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.annotation.PreDestroy;
+
+import org.lucares.pdb.api.Entry;
+import org.lucares.pdb.api.Tags;
+import org.lucares.performance.db.BlockingQueueIterator;
+import org.lucares.performance.db.PerformanceDb;
+import org.lucares.recommind.logs.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+
+@Component
+public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
+	private static final Logger LOGGER = LoggerFactory.getLogger(TcpIngestor.class);
+	private static final Logger METRICS_LOGGER = LoggerFactory.getLogger("org.lucares.metrics.tcpIngestor");
+
+	public static final int PORT = 17347;
+
+	private final AtomicBoolean acceptNewConnections = new AtomicBoolean(true);
+
+	private final ExecutorService serverThreadPool = Executors.newFixedThreadPool(2);
+
+	private final ExecutorService workerThreadPool = Executors.newCachedThreadPool();
+
+	private final PerformanceDb db;
+
+	public final static class Handler implements Callable<Void> {
+
+		private final ObjectMapper objectMapper = new ObjectMapper();
+
+		private final TypeReference<Map<String, Object>> typeReferenceForMap = new TypeReference<Map<String, Object>>() {
+		};
+
+		final Socket clientSocket;
+		private final ArrayBlockingQueue<Entry> queue;
+
+		public Handler(final Socket clientSocket, final ArrayBlockingQueue<Entry> queue) {
+			this.clientSocket = clientSocket;
+			this.queue = queue;
+		}
+
+		@Override
+		public Void call() throws Exception {
+			final SocketAddress clientAddress = clientSocket.getRemoteSocketAddress();
+			Thread.currentThread().setName("worker-" + clientAddress);
+			LOGGER.debug("opening streams to client");
+			try (PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);
+					BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+
+			) {
+				final ObjectReader objectReader = objectMapper.readerFor(typeReferenceForMap);
+				final MappingIterator<Object> iterator = objectReader.readValues(in);
+
+				double duration = 0.0;
+				int count = 0;
+				LOGGER.debug("reading from stream");
+				while (iterator.hasNext()) {
+
+					final long start = System.nanoTime();
+					@SuppressWarnings("unchecked")
+					final Map<String, Object> object = (Map<String, Object>) iterator.next();
+
+					final Optional<Entry> entry = createEntry(object);
+					final long end = System.nanoTime();
+					duration += (end - start) / 1_000_000.0;
+
+					count++;
+					if (count == 100000) {
+						METRICS_LOGGER.debug("reading {} took {} ms", count, duration);
+						duration = 0.0;
+						count = 0;
+					}
+
+					if (entry.isPresent()) {
+						LOGGER.trace("adding entry to queue: {}", entry);
+						queue.put(entry.get());
+					}
+
+				}
+				LOGGER.debug("connection closed: " + clientAddress);
+			}
+
+			return null;
+		}
+
+		public Optional<Entry> createEntry(final Map<String, Object> map) {
+			try {
+
+				final OffsetDateTime date = getDate(map);
+				final long duration = (int) map.get("duration");
+
+				final Tags tags = createTags(map);
+
+				final Entry entry = new Entry(date, duration, tags);
+				return Optional.of(entry);
+			} catch (final Exception e) {
+				return Optional.empty();
+			}
+		}
+
+		private Tags createTags(final Map<String, Object> map) {
+			Tags tags = Tags.create();
+			for (final java.util.Map.Entry<String, Object> e : map.entrySet()) {
+
+				final String key = e.getKey();
+				final Object value = e.getValue();
+
+				switch (key) {
+				case "@timestamp":
+				case "duration":
+					// these fields are not tags
+					break;
+				case "tags":
+					// TODO @ahr add support for simple tags, currently we
+					// only support key/value tags
+					break;
+				default:
+					tags = tags.copyAddIfNotNull(key, String.valueOf(value));
+					break;
+				}
+			}
+			return tags;
+		}
+
+		private OffsetDateTime getDate(final Map<String, Object> map) {
+			final String timestamp = (String) map.get("@timestamp");
+
+			final OffsetDateTime date = OffsetDateTime.parse(timestamp, DateTimeFormatter.ISO_ZONED_DATE_TIME);
+			return date;
+		}
+	}
+
+	public TcpIngestor(final Path dataDirectory) {
+		LOGGER.info("opening performance db: " + dataDirectory);
+		db = new PerformanceDb(dataDirectory);
+		LOGGER.debug("performance db open");
+	}
+
+	@Autowired
+	public TcpIngestor(final PdbRepository pdbRepository) {
+		db = pdbRepository.getDb();
+	}
+
+	@Async
+	@Override
+	public void start() throws Exception {
+
+		final ArrayBlockingQueue<Entry> queue = new ArrayBlockingQueue<>(1);
+
+		serverThreadPool.submit(() -> {
+			Thread.currentThread().setName("db-ingestion");
+
+			boolean finished = false;
+			while (!finished) {
+				try {
+					db.put(new BlockingQueueIterator<>(queue, Entry.POISON));
+					finished = true;
+				} catch (final Exception e) {
+					LOGGER.warn("Write to database failed. Will retry with the next element.", e);
+				}
+			}
+			return null;
+		});
+
+		serverThreadPool.submit(() -> listen(queue));
+	}
+
+	private Void listen(final ArrayBlockingQueue<Entry> queue) throws IOException {
+		Thread.currentThread().setName("socket-listener");
+		try (ServerSocket serverSocket = new ServerSocket(PORT);) {
+			LOGGER.info("listening on port " + PORT);
+
+			serverSocket.setSoTimeout((int) TimeUnit.MILLISECONDS.toMillis(100));
+
+			while (acceptNewConnections.get()) {
+				try {
+					final Socket clientSocket = serverSocket.accept();
+					LOGGER.debug("accepted connection: " + clientSocket.getRemoteSocketAddress());
+
+					workerThreadPool.submit(new Handler(clientSocket, queue));
+					LOGGER.debug("handler submitted");
+				} catch (final SocketTimeoutException e) {
+					// expected every 100ms
+					// needed to be able to stop the server
+				} catch (final Exception e) {
+					LOGGER.warn("Exception caught while waiting for a new connection. "
+							+ "We'll ignore this error and keep going.", e);
+				}
+			}
+			LOGGER.info("not accepting new connections. ");
+
+			LOGGER.info("stopping worker pool");
+			workerThreadPool.shutdown();
+			try {
+				workerThreadPool.awaitTermination(10, TimeUnit.MINUTES);
+				LOGGER.debug("workers stopped");
+			} catch (final InterruptedException e) {
+				Thread.interrupted();
+			}
+			LOGGER.debug("adding poison");
+			queue.put(Entry.POISON);
+		} catch (final InterruptedException e) {
+			LOGGER.info("Listener thread interrupted. Likely while adding the poison. "
+					+ "That would mean that the db-ingestion thread will not terminate. ");
+			Thread.interrupted();
+		} catch (final Exception e) {
+			LOGGER.error("", e);
+			throw e;
+		}
+		return null;
+	}
+
+	@Override
+	@PreDestroy
+	public void destroy() {
+		close();
+	}
+
+	@Override
+	public void close() {
+		LOGGER.debug("stopping accept thread");
+		acceptNewConnections.set(false);
+		serverThreadPool.shutdown();
+		try {
+			serverThreadPool.awaitTermination(10, TimeUnit.MINUTES);
+		} catch (final InterruptedException e) {
+			Thread.interrupted();
+		}
+		LOGGER.debug("closing database");
+		db.close();
+		LOGGER.info("destroyed");
+	}
+
+	public static void main(final String[] args) throws Exception {
+
+		Runtime.getRuntime().addShutdownHook(new Thread() {
+			@Override
+			public void run() {
+				LOGGER.info("shutdown hook");
+			}
+		});
+
+		try (final TcpIngestor ingestor = new TcpIngestor(Config.DATA_DIR)) {
+			ingestor.start();
+			TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);
+		}
+	}
+}

pdb-ui/src/main/resources/application-dev.properties

@@ -0,0 +1 @@
+db.base=/home/andi/ws/performanceDb/db

pdb-ui/src/main/resources/application.properties

@@ -0,0 +1,4 @@
+db.base=${java.io.tmpdir}/db
+
+path.tmp=${java.io.tmpdir}/pdb/tmp
+path.output=${java.io.tmpdir}/pdb/out

pdb-ui/src/main/resources/config.system.properties

@@ -1,4 +0,0 @@
-db.base=/tmp/db
-
-path.tmp=/tmp/pdb/tmp
-path.output=/tmp/pdb/out

pdb-ui/src/test/java/org/lucares/performance/db/ingestor/TcpIngestorTest.java

@@ -0,0 +1,171 @@
+package org.lucares.performance.db.ingestor;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.lucares.pdb.api.Entry;
+import org.lucares.pdbui.TcpIngestor;
+import org.lucares.performance.db.FileUtils;
+import org.lucares.performance.db.PerformanceDb;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import liquibase.exception.LiquibaseException;
+
+@Test
+public class TcpIngestorTest {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(TcpIngestorTest.class);
+
+	private Path dataDirectory;
+
+	@BeforeMethod
+	public void beforeMethod() throws IOException {
+		dataDirectory = Files.createTempDirectory("pdb");
+	}
+
+	@AfterMethod
+	public void afterMethod() throws IOException {
+		FileUtils.delete(dataDirectory);
+	}
+
+	public void testIngestDataViaTcpStream() throws LiquibaseException, Exception {
+
+		final OffsetDateTime dateA = OffsetDateTime.now();
+		final OffsetDateTime dateB = OffsetDateTime.now();
+		final String host = "someHost";
+
+		try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
+
+			ingestor.start();
+
+			final Map<String, Object> entryA = new HashMap<>();
+			entryA.put("duration", 1);
+			entryA.put("@timestamp", dateA.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
+			entryA.put("host", host);
+			entryA.put("tags", Collections.emptyList());
+
+			final Map<String, Object> entryB = new HashMap<>();
+			entryB.put("duration", 2);
+			entryB.put("@timestamp", dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
+			entryB.put("host", host);
+			entryB.put("tags", Collections.emptyList());
+
+			send(entryA, entryB);
+		} catch (final Exception e) {
+			LOGGER.error("", e);
+			throw e;
+		}
+
+		try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
+			final List<Entry> result = db.get("host=" + host).singleGroup().asList();
+			Assert.assertEquals(result.size(), 2);
+
+			Assert.assertEquals(result.get(0).getValue(), 1);
+			Assert.assertEquals(result.get(0).getDate().toInstant(), dateA.toInstant());
+
+			Assert.assertEquals(result.get(1).getValue(), 2);
+			Assert.assertEquals(result.get(1).getDate().toInstant(), dateB.toInstant());
+		}
+	}
+
+	@SafeVarargs
+	private final void send(final Map<String, Object>... entries) throws IOException {
+
+		final StringBuilder streamData = new StringBuilder();
+		final ObjectMapper mapper = new ObjectMapper();
+
+		for (final Map<String, Object> entry : entries) {
+
+			streamData.append(mapper.writeValueAsString(entry));
+			streamData.append("\n");
+		}
+
+		final SocketChannel channel = connect();
+		final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8));
+		channel.write(src);
+
+		try {
+			// ugly workaround: the channel was close too early and not all data
+			// was received
+			TimeUnit.MILLISECONDS.sleep(10);
+		} catch (final InterruptedException e) {
+			throw new IllegalStateException(e);
+		}
+		channel.close();
+		LOGGER.trace("closed sender connection");
+	}
+
+	private SocketChannel connect() throws IOException {
+
+		SocketChannel result = null;
+
+		while (true) {
+			try {
+				result = SocketChannel.open();
+				result.configureBlocking(true);
+				result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT));
+				break;
+			} catch (final ConnectException e) {
+				// server socket not yet ready, it should be ready any time soon
+			}
+		}
+
+		return result;
+	}
+
+	@Test
+	public void testIngestionThreadDoesNotDieOnErrors() throws Exception {
+		final OffsetDateTime invalidDate = OffsetDateTime.ofInstant(Instant.ofEpochMilli(-1), ZoneOffset.UTC);
+		final OffsetDateTime dateB = OffsetDateTime.now();
+		final String host = "someHost";
+
+		try (TcpIngestor tcpIngestor = new TcpIngestor(dataDirectory)) {
+			tcpIngestor.start();
+
+			// this entry will be skipped, because the date is invalid
+			final Map<String, Object> entryA = new HashMap<>();
+			entryA.put("duration", 1);
+			entryA.put("@timestamp", invalidDate.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
+			entryA.put("host", host);
+			entryA.put("tags", Collections.emptyList());
+
+			final Map<String, Object> entryB = new HashMap<>();
+			entryB.put("duration", 2);
+			entryB.put("@timestamp", dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
+			entryB.put("host", host);
+			entryB.put("tags", Collections.emptyList());
+
+			send(entryA, entryB);
+		}
+
+		try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
+			final List<Entry> result = db.get("host=" + host).singleGroup().asList();
+			Assert.assertEquals(result.size(), 1);
+
+			Assert.assertEquals(result.get(0).getValue(), 2);
+			Assert.assertEquals(result.get(0).getDate().toInstant(), dateB.toInstant());
+		}
+	}
+}