mirror of
https://github.com/NationalSecurityAgency/timely.git
synced 2026-11-09 17:36:01 -10:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Bill Oley
|
eaddee0d78 | ||
|
Bill Oley
|
e947c1a497 | ||
|
Bill Oley
|
4e7e07601b | ||
|
Bill Oley
|
679a92c5b5 |
No files matched your search
@@ -6,6 +6,4 @@ target/
|
||||
*.iml
|
||||
logs/
|
||||
server/pki/
|
||||
client/src/main/python/setup.py
|
||||
grafana/timely-app/yarn-error.log
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
|
||||
Apache License
|
||||
Version 2.0, January 2004
|
||||
httpProperties://www.apache.org/licenses/
|
||||
http://www.apache.org/licenses/
|
||||
|
||||
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
|
||||
|
||||
@@ -193,7 +193,7 @@
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
httpProperties://www.apache.org/licenses/LICENSE-2.0
|
||||
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,
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
This product includes Google Flatbuffers (https://google.github.io/flatbuffers/)
|
||||
Copyright 2014 Google Inc. All rights reserved.
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
[![Apache License][li]][ll]
|
||||
|
||||
Timely is a time series database application that provides secure access to time series data. Timely is written in Java and designed to work with [Apache Accumulo](https://accumuloProperties.apache.org/) and [Grafana](https://www.grafana.com). Documentation is located [here](https://nationalsecurityagency.github.io/timely/).
|
||||
Timely is a time series database application that provides secure access to time series data. Timely is written in Java and designed to work with [Apache Accumulo](http://accumulo.apache.org/) and [Grafana](http://www.grafana.org). Documentation is located [here](https://nationalsecurityagency.github.io/timely/).
|
||||
|
||||
[li]: https://img.shields.io/badge/license-ASL-blue.svg
|
||||
[li]: http://img.shields.io/badge/license-ASL-blue.svg
|
||||
[ll]: https://www.apache.org/licenses/LICENSE-2.0
|
||||
+321
-255
@@ -1,270 +1,336 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<?xml version="1.0"?>
|
||||
<!--
|
||||
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.
|
||||
-->
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely</artifactId>
|
||||
<version>2.1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>timely-analytics</artifactId>
|
||||
<!--
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>timely-analytics</artifactId>
|
||||
<parent>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely</artifactId>
|
||||
<version>0.0.6-SNAPSHOT</version>
|
||||
</parent>
|
||||
<properties>
|
||||
<flink.version>1.1.4</flink.version>
|
||||
</properties>
|
||||
<!--
|
||||
|
||||
Execute "mvn clean package -Pbuild-jar"
|
||||
to build a jar file out of this project!
|
||||
|
||||
Execute "mvn clean package -Pbuild-jar"
|
||||
to build a jar file out of this project!
|
||||
How to use the Flink Quickstart pom:
|
||||
|
||||
How to use the Flink Quickstart pom:
|
||||
a) Adding new dependencies:
|
||||
You can add dependencies to the list below.
|
||||
Please check if the maven-shade-plugin below is filtering out your dependency
|
||||
and remove the exclude from there.
|
||||
|
||||
a) Adding new dependencies:
|
||||
You can add dependencies to the list below.
|
||||
Please check if the maven-shade-plugin below is filtering out your dependency
|
||||
and remove the exclude from there.
|
||||
b) Build a jar for running on the cluster:
|
||||
There are two options for creating a jar from this project
|
||||
|
||||
b) Build a jar for running on the cluster:
|
||||
There are two options for creating a jar from this project
|
||||
b.1) "mvn clean package" -> this will create a fat jar which contains all
|
||||
dependencies necessary for running the jar created by this pom in a cluster.
|
||||
The "maven-shade-plugin" excludes everything that is provided on a running Flink cluster.
|
||||
|
||||
b.1) "mvn clean package" -> this will create a fat jar which contains all
|
||||
dependencies necessary for running the jar created by this pom in a cluster.
|
||||
The "maven-shade-plugin" excludes everything that is provided on a running Flink cluster.
|
||||
|
||||
b.2) "mvn clean package -Pbuild-jar" -> This will also create a fat-jar, but with much
|
||||
nicer dependency exclusion handling. This approach is preferred and leads to
|
||||
much cleaner jar files.
|
||||
-->
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-clients_2.11</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-java</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-streaming-java_2.11</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-client</artifactId>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>*</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-common</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<pluginManagement>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<compilerId>jdt</compilerId>
|
||||
</configuration>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.eclipse.tycho</groupId>
|
||||
<artifactId>tycho-compiler-jdt</artifactId>
|
||||
<version>1.4.0</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</pluginManagement>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.jacoco</groupId>
|
||||
<artifactId>jacoco-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<!-- We use the maven-shade plugin to create a fat jar that contains all dependencies
|
||||
except flink and it's transitive dependencies. The resulting fat-jar can be executed
|
||||
on a cluster. Change the value of Program-Class if your program entry point changes. -->
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-shade-plugin</artifactId>
|
||||
<executions>
|
||||
<!-- Run shade goal on package phase -->
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>shade</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<artifactSet>
|
||||
<excludes>
|
||||
<!-- This list contains all dependencies of flink-dist
|
||||
Everything else will be packaged into the fat-jar
|
||||
-->
|
||||
<exclude>org.apache.flink:flink-annotations</exclude>
|
||||
<exclude>org.apache.flink:flink-shaded-hadoop1_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
|
||||
<exclude>org.apache.flink:flink-shaded-curator-recipes</exclude>
|
||||
<exclude>org.apache.flink:flink-core</exclude>
|
||||
<exclude>org.apache.flink:flink-java</exclude>
|
||||
<exclude>org.apache.flink:flink-scala_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-runtime_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-optimizer_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-clients_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-avro_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-examples-batch_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-examples-streaming_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-streaming-java_2.11</exclude>
|
||||
<!-- Also exclude very big transitive dependencies of Flink
|
||||
WARNING: You have to remove these excludes if your code relies on other
|
||||
versions of these dependencies.
|
||||
-->
|
||||
<exclude>org.scala-lang:scala-library</exclude>
|
||||
<exclude>org.scala-lang:scala-compiler</exclude>
|
||||
<exclude>org.scala-lang:scala-reflect</exclude>
|
||||
<exclude>com.typesafe.akka:akka-actor_*</exclude>
|
||||
<exclude>com.typesafe.akka:akka-remote_*</exclude>
|
||||
<exclude>com.typesafe.akka:akka-slf4j_*</exclude>
|
||||
<exclude>io.netty:netty-all</exclude>
|
||||
<exclude>io.netty:netty</exclude>
|
||||
<exclude>commons-fileupload:commons-fileupload</exclude>
|
||||
<exclude>org.apache.avro:avro</exclude>
|
||||
<exclude>commons-collections:commons-collections</exclude>
|
||||
<exclude>org.codehaus.jackson:jackson-core-asl</exclude>
|
||||
<exclude>org.codehaus.jackson:jackson-mapper-asl</exclude>
|
||||
<exclude>com.thoughtworks.paranamer:paranamer</exclude>
|
||||
<exclude>org.xerial.snappy:snappy-java</exclude>
|
||||
<exclude>org.apache.commons:commons-compress</exclude>
|
||||
<exclude>org.tukaani:xz</exclude>
|
||||
<exclude>com.esotericsoftware.kryo:kryo</exclude>
|
||||
<exclude>com.esotericsoftware.minlog:minlog</exclude>
|
||||
<exclude>org.objenesis:objenesis</exclude>
|
||||
<exclude>com.twitter:chill_*</exclude>
|
||||
<exclude>com.twitter:chill-java</exclude>
|
||||
<exclude>commons-lang:commons-lang</exclude>
|
||||
<exclude>junit:junit</exclude>
|
||||
<exclude>org.apache.commons:commons-lang3</exclude>
|
||||
<exclude>org.slf4j:slf4j-api</exclude>
|
||||
<exclude>org.slf4j:slf4j-log4j12</exclude>
|
||||
<exclude>log4j:log4j</exclude>
|
||||
<exclude>org.apache.commons:commons-math</exclude>
|
||||
<exclude>org.apache.sling:org.apache.sling.commons.json</exclude>
|
||||
<exclude>commons-logging:commons-logging</exclude>
|
||||
<exclude>commons-codec:commons-codec</exclude>
|
||||
<exclude>com.fasterxml.jackson.core:jackson-core</exclude>
|
||||
<exclude>com.fasterxml.jackson.core:jackson-databind</exclude>
|
||||
<exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
|
||||
<exclude>stax:stax-api</exclude>
|
||||
<exclude>com.typesafe:config</exclude>
|
||||
<exclude>org.uncommons.maths:uncommons-maths</exclude>
|
||||
<exclude>com.github.scopt:scopt_*</exclude>
|
||||
<exclude>commons-io:commons-io</exclude>
|
||||
<exclude>commons-cli:commons-cli</exclude>
|
||||
</excludes>
|
||||
</artifactSet>
|
||||
<filters>
|
||||
<filter>
|
||||
<artifact>*:*</artifact>
|
||||
<excludes>
|
||||
<exclude>**/module-info.class</exclude>
|
||||
</excludes>
|
||||
</filter>
|
||||
<filter>
|
||||
<artifact>org.apache.flink:*</artifact>
|
||||
<excludes>
|
||||
<!-- exclude shaded google but include shaded curator -->
|
||||
<exclude>org/apache/flink/shaded/com/**</exclude>
|
||||
<exclude>web-docs/**</exclude>
|
||||
</excludes>
|
||||
</filter>
|
||||
<filter>
|
||||
<!-- Do not copy the signatures in the META-INF folder.
|
||||
Otherwise, this might cause SecurityExceptions when using the JAR.
|
||||
-->
|
||||
<artifact>*:*</artifact>
|
||||
<excludes>
|
||||
<exclude>META-INF/*.SF</exclude>
|
||||
<exclude>META-INF/*.DSA</exclude>
|
||||
<exclude>META-INF/*.RSA</exclude>
|
||||
</excludes>
|
||||
</filter>
|
||||
</filters>
|
||||
<!-- If you want to use ./bin/flink run <quickstart jar> uncomment the following lines.
|
||||
This will add a Main-Class entry to the manifest file -->
|
||||
<!--
|
||||
<transformers>
|
||||
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
|
||||
<mainClass>jar.StreamingJob</mainClass>
|
||||
</transformer>
|
||||
</transformers>
|
||||
-->
|
||||
<createDependencyReducedPom>false</createDependencyReducedPom>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>buildnumber-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
<profiles>
|
||||
<profile>
|
||||
<!-- Profile for packaging correct JAR files -->
|
||||
<id>build-jar</id>
|
||||
<dependencies>
|
||||
b.2) "mvn clean package -Pbuild-jar" -> This will also create a fat-jar, but with much
|
||||
nicer dependency exclusion handling. This approach is preferred and leads to
|
||||
much cleaner jar files.
|
||||
-->
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-clients_2.11</artifactId>
|
||||
<scope>provided</scope>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-java</artifactId>
|
||||
<scope>provided</scope>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-java</artifactId>
|
||||
<version>${flink.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-streaming-java_2.11</artifactId>
|
||||
<scope>provided</scope>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-streaming-java_2.11</artifactId>
|
||||
<version>${flink.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-clients_2.11</artifactId>
|
||||
<version>${flink.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-client</artifactId>
|
||||
<version>${parent.version}</version>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>*</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-common</artifactId>
|
||||
<version>${parent.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<profiles>
|
||||
<profile>
|
||||
<!-- Profile for packaging correct JAR files -->
|
||||
<id>build-jar</id>
|
||||
<activation>
|
||||
<activeByDefault>false</activeByDefault>
|
||||
</activation>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-java</artifactId>
|
||||
<version>${flink.version}</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-streaming-java_2.11</artifactId>
|
||||
<version>${flink.version}</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.flink</groupId>
|
||||
<artifactId>flink-clients_2.11</artifactId>
|
||||
<version>${flink.version}</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<!-- disable the exclusion rules -->
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-shade-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<phase>package</phase>
|
||||
<goals>
|
||||
<goal>shade</goal>
|
||||
</goals>
|
||||
<configuration>
|
||||
<artifactSet>
|
||||
<excludes combine.self="override"/>
|
||||
</artifactSet>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</profile>
|
||||
</profiles>
|
||||
<build>
|
||||
<plugins>
|
||||
<!-- disable the exclusion rules -->
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-shade-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>shade</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<artifactSet>
|
||||
<excludes combine.self="override" />
|
||||
</artifactSet>
|
||||
<filters>
|
||||
<filter>
|
||||
<artifact>*:*</artifact>
|
||||
<excludes>
|
||||
<exclude>**/module-info.class</exclude>
|
||||
</excludes>
|
||||
</filter>
|
||||
</filters>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>net.revelc.code</groupId>
|
||||
<artifactId>formatter-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.jacoco</groupId>
|
||||
<artifactId>jacoco-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<!-- We use the maven-shade plugin to create a fat jar that contains all dependencies
|
||||
except flink and it's transitive dependencies. The resulting fat-jar can be executed
|
||||
on a cluster. Change the value of Program-Class if your program entry point changes. -->
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-shade-plugin</artifactId>
|
||||
<executions>
|
||||
<!-- Run shade goal on package phase -->
|
||||
<execution>
|
||||
<phase>package</phase>
|
||||
<goals>
|
||||
<goal>shade</goal>
|
||||
</goals>
|
||||
<configuration>
|
||||
<artifactSet>
|
||||
<excludes>
|
||||
<!-- This list contains all dependencies of flink-dist
|
||||
Everything else will be packaged into the fat-jar
|
||||
-->
|
||||
<exclude>org.apache.flink:flink-annotations</exclude>
|
||||
<exclude>org.apache.flink:flink-shaded-hadoop1_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
|
||||
<exclude>org.apache.flink:flink-shaded-curator-recipes</exclude>
|
||||
<exclude>org.apache.flink:flink-core</exclude>
|
||||
<exclude>org.apache.flink:flink-java</exclude>
|
||||
<exclude>org.apache.flink:flink-scala_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-runtime_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-optimizer_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-clients_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-avro_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-examples-batch_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-examples-streaming_2.11</exclude>
|
||||
<exclude>org.apache.flink:flink-streaming-java_2.11</exclude>
|
||||
<!-- Also exclude very big transitive dependencies of Flink
|
||||
WARNING: You have to remove these excludes if your code relies on other
|
||||
versions of these dependencies.
|
||||
-->
|
||||
<exclude>org.scala-lang:scala-library</exclude>
|
||||
<exclude>org.scala-lang:scala-compiler</exclude>
|
||||
<exclude>org.scala-lang:scala-reflect</exclude>
|
||||
<exclude>com.typesafe.akka:akka-actor_*</exclude>
|
||||
<exclude>com.typesafe.akka:akka-remote_*</exclude>
|
||||
<exclude>com.typesafe.akka:akka-slf4j_*</exclude>
|
||||
<exclude>io.netty:netty-all</exclude>
|
||||
<exclude>io.netty:netty</exclude>
|
||||
<exclude>commons-fileupload:commons-fileupload</exclude>
|
||||
<exclude>org.apache.avro:avro</exclude>
|
||||
<exclude>commons-collections:commons-collections</exclude>
|
||||
<exclude>org.codehaus.jackson:jackson-core-asl</exclude>
|
||||
<exclude>org.codehaus.jackson:jackson-mapper-asl</exclude>
|
||||
<exclude>com.thoughtworks.paranamer:paranamer</exclude>
|
||||
<exclude>org.xerial.snappy:snappy-java</exclude>
|
||||
<exclude>org.apache.commons:commons-compress</exclude>
|
||||
<exclude>org.tukaani:xz</exclude>
|
||||
<exclude>com.esotericsoftware.kryo:kryo</exclude>
|
||||
<exclude>com.esotericsoftware.minlog:minlog</exclude>
|
||||
<exclude>org.objenesis:objenesis</exclude>
|
||||
<exclude>com.twitter:chill_*</exclude>
|
||||
<exclude>com.twitter:chill-java</exclude>
|
||||
<exclude>commons-lang:commons-lang</exclude>
|
||||
<exclude>junit:junit</exclude>
|
||||
<exclude>org.apache.commons:commons-lang3</exclude>
|
||||
<exclude>org.slf4j:slf4j-api</exclude>
|
||||
<exclude>org.slf4j:slf4j-log4j12</exclude>
|
||||
<exclude>log4j:log4j</exclude>
|
||||
<exclude>org.apache.commons:commons-math</exclude>
|
||||
<exclude>org.apache.sling:org.apache.sling.commons.json</exclude>
|
||||
<exclude>commons-logging:commons-logging</exclude>
|
||||
<exclude>commons-codec:commons-codec</exclude>
|
||||
<exclude>com.fasterxml.jackson.core:jackson-core</exclude>
|
||||
<exclude>com.fasterxml.jackson.core:jackson-databind</exclude>
|
||||
<exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
|
||||
<exclude>stax:stax-api</exclude>
|
||||
<exclude>com.typesafe:config</exclude>
|
||||
<exclude>org.uncommons.maths:uncommons-maths</exclude>
|
||||
<exclude>com.github.scopt:scopt_*</exclude>
|
||||
<exclude>commons-io:commons-io</exclude>
|
||||
<exclude>commons-cli:commons-cli</exclude>
|
||||
</excludes>
|
||||
</artifactSet>
|
||||
<filters>
|
||||
<filter>
|
||||
<artifact>org.apache.flink:*</artifact>
|
||||
<excludes>
|
||||
<!-- exclude shaded google but include shaded curator -->
|
||||
<exclude>org/apache/flink/shaded/com/**</exclude>
|
||||
<exclude>web-docs/**</exclude>
|
||||
</excludes>
|
||||
</filter>
|
||||
<filter>
|
||||
<!-- Do not copy the signatures in the META-INF folder.
|
||||
Otherwise, this might cause SecurityExceptions when using the JAR.
|
||||
-->
|
||||
<artifact>*:*</artifact>
|
||||
<excludes>
|
||||
<exclude>META-INF/*.SF</exclude>
|
||||
<exclude>META-INF/*.DSA</exclude>
|
||||
<exclude>META-INF/*.RSA</exclude>
|
||||
</excludes>
|
||||
</filter>
|
||||
</filters>
|
||||
<!-- If you want to use ./bin/flink run <quickstart jar> uncomment the following lines.
|
||||
This will add a Main-Class entry to the manifest file -->
|
||||
<!--
|
||||
<transformers>
|
||||
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
|
||||
<mainClass>jar.StreamingJob</mainClass>
|
||||
</transformer>
|
||||
</transformers>
|
||||
-->
|
||||
<createDependencyReducedPom>false</createDependencyReducedPom>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>buildnumber-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</profile>
|
||||
</profiles>
|
||||
<pluginManagement>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>1.8</source>
|
||||
<target>1.8</target>
|
||||
<compilerId>jdt</compilerId>
|
||||
</configuration>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.eclipse.tycho</groupId>
|
||||
<artifactId>tycho-compiler-jdt</artifactId>
|
||||
<version>0.21.0</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.eclipse.m2e</groupId>
|
||||
<artifactId>lifecycle-mapping</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<configuration>
|
||||
<lifecycleMappingMetadata>
|
||||
<pluginExecutions>
|
||||
<pluginExecution>
|
||||
<pluginExecutionFilter>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-assembly-plugin</artifactId>
|
||||
<versionRange>[2.4,)</versionRange>
|
||||
<goals>
|
||||
<goal>single</goal>
|
||||
</goals>
|
||||
</pluginExecutionFilter>
|
||||
<action>
|
||||
<ignore/>
|
||||
</action>
|
||||
</pluginExecution>
|
||||
<pluginExecution>
|
||||
<pluginExecutionFilter>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<versionRange>[3.1,)</versionRange>
|
||||
<goals>
|
||||
<goal>testCompile</goal>
|
||||
<goal>compile</goal>
|
||||
</goals>
|
||||
</pluginExecutionFilter>
|
||||
<action>
|
||||
<ignore/>
|
||||
</action>
|
||||
</pluginExecution>
|
||||
</pluginExecutions>
|
||||
</lifecycleMappingMetadata>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</pluginManagement>
|
||||
</build>
|
||||
</project>
|
||||
@@ -5,9 +5,10 @@ import org.apache.flink.api.java.functions.KeySelector;
|
||||
import timely.api.response.MetricResponse;
|
||||
|
||||
/**
|
||||
* Uniquely defines a time series in a Flink stream. A unique time series is the metric name and all of its tags, to include visibility if it exists.
|
||||
* Uniquely defines a time series in a Flink stream. A unique time series is the
|
||||
* metric name and all of its tags, to include visibility if it exists.
|
||||
*/
|
||||
public class MetricKeySelector implements KeySelector<MetricResponse,String> {
|
||||
public class MetricKeySelector implements KeySelector<MetricResponse, String> {
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
|
||||
@@ -9,11 +9,11 @@ import org.apache.flink.api.common.accumulators.Accumulator;
|
||||
* Custom accumulator that retains a sorted map of strings and their counts
|
||||
*
|
||||
*/
|
||||
public class SortedStringAccumulator implements Accumulator<String,ConcurrentSkipListMap<String,MutableLong>> {
|
||||
public class SortedStringAccumulator implements Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> {
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private ConcurrentSkipListMap<String,MutableLong> values = new ConcurrentSkipListMap<>();
|
||||
private ConcurrentSkipListMap<String, MutableLong> values = new ConcurrentSkipListMap<>();
|
||||
|
||||
@Override
|
||||
public void add(String value) {
|
||||
@@ -25,7 +25,7 @@ public class SortedStringAccumulator implements Accumulator<String,ConcurrentSki
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConcurrentSkipListMap<String,MutableLong> getLocalValue() {
|
||||
public ConcurrentSkipListMap<String, MutableLong> getLocalValue() {
|
||||
return values;
|
||||
}
|
||||
|
||||
@@ -35,7 +35,7 @@ public class SortedStringAccumulator implements Accumulator<String,ConcurrentSki
|
||||
}
|
||||
|
||||
@Override
|
||||
public void merge(Accumulator<String,ConcurrentSkipListMap<String,MutableLong>> other) {
|
||||
public void merge(Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> other) {
|
||||
other.getLocalValue().forEach((k, v) -> {
|
||||
if (values.containsKey(k)) {
|
||||
values.get(k).add(v.longValue());
|
||||
@@ -46,7 +46,7 @@ public class SortedStringAccumulator implements Accumulator<String,ConcurrentSki
|
||||
}
|
||||
|
||||
@Override
|
||||
public Accumulator<String,ConcurrentSkipListMap<String,MutableLong>> clone() {
|
||||
public Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> clone() {
|
||||
SortedStringAccumulator a = new SortedStringAccumulator();
|
||||
a.getLocalValue().putAll(this.values);
|
||||
return a;
|
||||
|
||||
@@ -42,7 +42,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
}
|
||||
}
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(SubscriptionSource.class);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionSource.class);
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private transient WebSocketSubscriptionClient client = null;
|
||||
@@ -64,11 +64,12 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
|
||||
@Override
|
||||
public void open(Configuration parameters) throws Exception {
|
||||
log.info("Opening summarization job.");
|
||||
LOG.info("Opening summarization job.");
|
||||
super.open(parameters);
|
||||
client = new WebSocketSubscriptionClient(jp.getTimelyHostname(), jp.getTimelyHttpsPort(), jp.getTimelyWssPort(), jp.isClientAuth(), jp.isDoLogin(),
|
||||
jp.getKeyStoreFile(), jp.getKeyStoreType(), jp.getKeyStorePass(), jp.getTrustStoreFile(), jp.getTrustStoreType(),
|
||||
jp.getTrustStorePass(), jp.isHostVerificationEnabled(), jp.getBufferSize());
|
||||
client = new WebSocketSubscriptionClient(jp.getTimelyHostname(), jp.getTimelyHttpsPort(),
|
||||
jp.getTimelyWssPort(), jp.isDoLogin(), jp.getTimelyUsername(), jp.getTimelyPassword(),
|
||||
jp.getKeyStoreFile(), jp.getKeyStoreType(), jp.getKeyStorePass(), jp.getTrustStoreFile(),
|
||||
jp.getTrustStoreType(), jp.getTrustStorePass(), jp.isHostVerificationEnabled(), jp.getBufferSize());
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -81,7 +82,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
|
||||
@Override
|
||||
public void run(SourceContext<MetricResponse> ctx) throws Exception {
|
||||
log.info("Running summarization job");
|
||||
LOG.info("Running summarization job");
|
||||
final SortedStringAccumulator dateTimeAccumulator = new SortedStringAccumulator();
|
||||
this.getRuntimeContext().addAccumulator("source hourly count", dateTimeAccumulator);
|
||||
final LongCounter sourceInputs = new LongCounter();
|
||||
@@ -96,40 +97,42 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
|
||||
@Override
|
||||
public void onOpen(Session session, EndpointConfig config) {
|
||||
session.addMessageHandler(String.class, new MessageHandler.Whole<String>() {
|
||||
session.addMessageHandler(new MessageHandler.Whole<String>() {
|
||||
|
||||
private final SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd-HH");
|
||||
private long lastWatermarkTime = 0L;
|
||||
|
||||
@Override
|
||||
public void onMessage(final String message) {
|
||||
log.info("Message received on Websocket session {}, length: {}", session.getId(), message.length());
|
||||
LOG.info("Message received on Websocket session {}, length: {}", session.getId(),
|
||||
message.length());
|
||||
try {
|
||||
// Deserialize in this thread
|
||||
final MetricResponses responses = JsonSerializer.getObjectMapper().readValue(message, MetricResponses.class);
|
||||
final MetricResponses responses = JsonSerializer.getObjectMapper().readValue(message,
|
||||
MetricResponses.class);
|
||||
// Process in the other
|
||||
final Runnable r = () -> {
|
||||
responses.getResponses().forEach(response -> {
|
||||
if (response.isComplete()) {
|
||||
log.info("Received last message.");
|
||||
LOG.info("Received last message.");
|
||||
ctx.emitWatermark(new Watermark(Long.MAX_VALUE));
|
||||
subscriptionsRemaining.getAndDecrement();
|
||||
return;
|
||||
}
|
||||
log.trace("Received metric: {}", response);
|
||||
LOG.trace("Received metric: {}", response);
|
||||
long time = response.getTimestamp();
|
||||
ctx.collectWithTimestamp(response, time);
|
||||
dateTimeAccumulator.add(formatter.format(new Date(response.getTimestamp())));
|
||||
sourceInputs.add(1);
|
||||
// Emit a watermark every second of event
|
||||
// time
|
||||
if (lastWatermarkTime == 0) {
|
||||
lastWatermarkTime = time;
|
||||
} else if ((time - lastWatermarkTime) > window) {
|
||||
lastWatermarkTime = time;
|
||||
ctx.emitWatermark(new Watermark(time - 1));
|
||||
}
|
||||
});
|
||||
if (lastWatermarkTime == 0) {
|
||||
lastWatermarkTime = time;
|
||||
} else if ((time - lastWatermarkTime) > window) {
|
||||
lastWatermarkTime = time;
|
||||
ctx.emitWatermark(new Watermark(time - 1));
|
||||
}
|
||||
});
|
||||
};
|
||||
try {
|
||||
svc.execute(r);
|
||||
@@ -139,11 +142,11 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
inputProcessedInMainThread.add(1);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
log.error("Error deserializing metric response, closing", e);
|
||||
LOG.error("Error deserializing metric response, closing", e);
|
||||
try {
|
||||
close();
|
||||
} catch (Exception e1) {
|
||||
log.error("Error closing client", e1);
|
||||
LOG.error("Error closing client", e1);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -156,7 +159,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
try {
|
||||
close();
|
||||
} catch (Exception e1) {
|
||||
log.error("Error closing client", e1);
|
||||
LOG.error("Error closing client", e1);
|
||||
}
|
||||
// Signal done sending data
|
||||
ctx.emitWatermark(new Watermark(Long.MAX_VALUE));
|
||||
@@ -168,7 +171,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
try {
|
||||
close();
|
||||
} catch (Exception e1) {
|
||||
log.error("Error closing client", e1);
|
||||
LOG.error("Error closing client", e1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -176,14 +179,14 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
try {
|
||||
client.open(handler);
|
||||
for (String m : metrics) {
|
||||
log.info("Adding subscription for {}", m);
|
||||
LOG.info("Adding subscription for {}", m);
|
||||
client.addSubscription(m, null, start, end, 5000);
|
||||
}
|
||||
while (!client.isClosed() && subscriptionsRemaining.get() > 0) {
|
||||
try {
|
||||
Thread.sleep(5000);
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Sleep interrupted.");
|
||||
LOG.error("Sleep interrupted.");
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
@@ -193,21 +196,21 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
log.info("Cancelling subscription source.");
|
||||
LOG.info("Cancelling subscription source.");
|
||||
try {
|
||||
close();
|
||||
} catch (Exception e) {
|
||||
log.error("Error closing web socket client");
|
||||
LOG.error("Error closing web socket client");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
log.info("Stopping subscription source.");
|
||||
LOG.info("Stopping subscription source.");
|
||||
try {
|
||||
close();
|
||||
} catch (Exception e) {
|
||||
log.error("Error closing web socket client");
|
||||
LOG.error("Error closing web socket client");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ import timely.api.response.MetricResponse;
|
||||
|
||||
public class SummarizationJob {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(SummarizationJob.class);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(SummarizationJob.class);
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
@@ -27,11 +27,11 @@ public class SummarizationJob {
|
||||
System.exit(1);
|
||||
}
|
||||
|
||||
log.info("Loading properties from: {}", args[0]);
|
||||
LOG.info("Loading properties from: {}", args[0]);
|
||||
ParameterTool params = ParameterTool.fromPropertiesFile(args[0]);
|
||||
|
||||
final SummarizationJobParameters jp = new SummarizationJobParameters(params);
|
||||
log.info("Properties: {}", jp);
|
||||
LOG.info("Properties: {}", jp);
|
||||
|
||||
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
|
||||
// @formatter:off
|
||||
@@ -43,7 +43,7 @@ public class SummarizationJob {
|
||||
.window(TumblingEventTimeWindows.of(jp.getSummarizationInterval()))
|
||||
.fold(new MetricHistogram(), new FoldFunction<MetricResponse, MetricHistogram>() {
|
||||
|
||||
private final Logger log = LoggerFactory
|
||||
private final Logger LOG = LoggerFactory
|
||||
.getLogger("timely.analytics.SummarizationJob.FoldFunction@");
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@@ -59,7 +59,7 @@ public class SummarizationJob {
|
||||
throw new RuntimeException("Tags don't not match - partitioning problem");
|
||||
}
|
||||
}
|
||||
log.trace("Updating histogram for {} with value {} at {} for metric {}", histo.getMetric()
|
||||
LOG.trace("Updating histogram for {} with value {} at {} for metric {}", histo.getMetric()
|
||||
+ histo.getTags(), metric.getValue(), metric.getTimestamp(), metric.getMetric()
|
||||
+ metric.getTags());
|
||||
histo.update(metric.getValue(), metric.getTimestamp());
|
||||
@@ -68,10 +68,10 @@ public class SummarizationJob {
|
||||
})
|
||||
.name("Metric Windows")
|
||||
.addSink(
|
||||
new SocketClientSink<MetricHistogram>(jp.getTimelyHostname(), jp.getTimelyTcpPort(),
|
||||
new SocketClientSink<MetricHistogram>(jp.getTimelyHostname(), jp.getTimelyTcpPort(),
|
||||
new MetricHistogram(), 2, true) {
|
||||
|
||||
private final Logger log = LoggerFactory
|
||||
private final Logger LOG = LoggerFactory
|
||||
.getLogger("timely.analytics.SummarizationJob.SocketClientSink");
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final LongCounter sinkCounter = new LongCounter();
|
||||
@@ -83,7 +83,7 @@ public class SummarizationJob {
|
||||
super.open(parameters);
|
||||
this.getRuntimeContext().addAccumulator("sink counter", this.sinkCounter);
|
||||
this.getRuntimeContext().addAccumulator("sink hourly count", dateTimeAccumulator);
|
||||
log.info("Sink opened");
|
||||
LOG.info("Sink opened");
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -97,7 +97,7 @@ public class SummarizationJob {
|
||||
@Override
|
||||
public void close() throws Exception {
|
||||
super.close();
|
||||
log.info("Sink closed");
|
||||
LOG.info("Sink closed");
|
||||
}
|
||||
|
||||
}).name("Timely Metric Sink");
|
||||
|
||||
@@ -18,7 +18,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
|
||||
private final int timelyTcpPort;
|
||||
private final int timelyHttpsPort;
|
||||
private final int timelyWssPort;
|
||||
private final boolean clientAuth;
|
||||
private final boolean doLogin;
|
||||
private final String timelyUsername;
|
||||
private final String timelyPassword;
|
||||
@@ -41,7 +40,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
|
||||
timelyTcpPort = params.getInt("timelyTcpPort", 4241);
|
||||
timelyHttpsPort = params.getInt("timelyHttpsPort", 4242);
|
||||
timelyWssPort = params.getInt("timelyWssPort", 4243);
|
||||
clientAuth = params.getBoolean("clientAuth", false);
|
||||
doLogin = params.getBoolean("doLogin", false);
|
||||
timelyUsername = params.get("timelyUsername", null);
|
||||
timelyPassword = params.get("timelyPassword", null);
|
||||
@@ -71,7 +69,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
|
||||
timelyTcpPort = conf.getInteger("timelyTcpPort", 4241);
|
||||
timelyHttpsPort = conf.getInteger("timelyHttpsPort", 4242);
|
||||
timelyWssPort = conf.getInteger("timelyWssPort", 4243);
|
||||
clientAuth = conf.getBoolean("clientAuth", false);
|
||||
doLogin = conf.getBoolean("doLogin", false);
|
||||
timelyUsername = conf.getString("timelyUsername", null);
|
||||
timelyPassword = conf.getString("timelyPassword", null);
|
||||
@@ -115,10 +112,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
|
||||
return timelyWssPort;
|
||||
}
|
||||
|
||||
public boolean isClientAuth() {
|
||||
return clientAuth;
|
||||
}
|
||||
|
||||
public boolean isDoLogin() {
|
||||
return doLogin;
|
||||
}
|
||||
@@ -184,8 +177,8 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String,String> toMap() {
|
||||
Map<String,String> m = new HashMap<>();
|
||||
public Map<String, String> toMap() {
|
||||
Map<String, String> m = new HashMap<>();
|
||||
m.put("timelyHostname", this.timelyHostname);
|
||||
m.put("timelyHttpsPort", Integer.toString(this.timelyHttpsPort));
|
||||
m.put("timelyWssPort", Integer.toString(this.timelyWssPort));
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<FindBugsFilter>
|
||||
<Match>
|
||||
<Class name="timely.analytics.flink.MetricHistogram" />
|
||||
<Method name="serialize" />
|
||||
<Bug pattern="DM_DEFAULT_ENCODING" />
|
||||
</Match>
|
||||
<Match>
|
||||
<Class name="timely.analytics.flink.SortedStringAccumulator" />
|
||||
<Method name="clone" />
|
||||
<Bug pattern="CN_IDIOM_NO_SUPER_CALL" />
|
||||
</Match>
|
||||
<Match>
|
||||
<Class name="timely.analytics.flink.SummarizationJobParameters" />
|
||||
<Method name="getMetrics" />
|
||||
<Bug pattern="EI_EXPOSE_REP" />
|
||||
</Match>
|
||||
<Match>
|
||||
<!-- these are probably avoidable, but needs further triage to fix -->
|
||||
<Bug pattern="EI_EXPOSE_REP" />
|
||||
</Match>
|
||||
<Match>
|
||||
<!-- these are probably avoidable, but needs further triage to fix -->
|
||||
<Bug pattern="EI_EXPOSE_REP2" />
|
||||
</Match>
|
||||
</FindBugsFilter>
|
||||
@@ -96,10 +96,12 @@ public class MetricHistogramTest {
|
||||
for (String put : puts.split("\n")) {
|
||||
Metric metric = metricParser.parse(put);
|
||||
Assert.assertEquals("sys.cpu.user_summarized", metric.getName());
|
||||
metric.getTags().forEach(t -> {
|
||||
Assert.assertTrue(t.equals(t1) || t.equals(t2) || t.equals(avg) || t.equals(min) || t.equals(max) || t.equals(sum) || t.equals(count)
|
||||
|| t.equals(p50) || t.equals(p75) || t.equals(p90) || t.equals(p99));
|
||||
});
|
||||
metric.getTags().forEach(
|
||||
t -> {
|
||||
Assert.assertTrue(t.equals(t1) || t.equals(t2) || t.equals(avg) || t.equals(min)
|
||||
|| t.equals(max) || t.equals(sum) || t.equals(count) || t.equals(p50) || t.equals(p75)
|
||||
|| t.equals(p90) || t.equals(p99));
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,40 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [[ `uname` == "Darwin" ]]; then
|
||||
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
|
||||
else
|
||||
THIS_SCRIPT=`readlink -f $0`
|
||||
fi
|
||||
|
||||
THIS_DIR="${THIS_SCRIPT%/*}"
|
||||
BASE_DIR=$(cd $THIS_DIR/.. && pwd)
|
||||
BIN_DIR=${BASE_DIR}/bin
|
||||
CONF_DIR=${BASE_DIR}/conf
|
||||
|
||||
. ${BIN_DIR}/timely-balancer-env.sh
|
||||
|
||||
for i in $(seq 1 $NUM_PROFILES); do
|
||||
if [ "$NUM_PROFILES" -eq "1" ]; then
|
||||
FILENAME=timely-balancer.yml
|
||||
else
|
||||
FILENAME=timely-balancer-${i}.yml
|
||||
fi
|
||||
|
||||
echo "creating configuration file ${CONF_DIR}/${FILENAME}"
|
||||
cp ${CONF_DIR}/timely-balancer-template.yml ${CONF_DIR}/${FILENAME}
|
||||
# if [ "$NUM_PROFILES" -eq "1" ]; then
|
||||
# sed -i "s/\${INSTANCE}//g" ${CONF_DIR}/${FILENAME}
|
||||
# else
|
||||
# sed -i "s/\${INSTANCE}/${i}/g" ${CONF_DIR}/${FILENAME}
|
||||
# fi
|
||||
MULTIPLE=$(($i - 1))
|
||||
TCP_PORT=$(($TCP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
|
||||
sed -i "s/\${TCP_PORT}/${TCP_PORT}/g" ${CONF_DIR}/${FILENAME}
|
||||
HTTP_PORT=$(($HTTP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
|
||||
sed -i "s/\${HTTP_PORT}/${HTTP_PORT}/g" ${CONF_DIR}/${FILENAME}
|
||||
WEBSOCKET_PORT=$(($WEBSOCKET_PORT_START + $MULTIPLE * $PORT_INCREMENT))
|
||||
sed -i "s/\${WEBSOCKET_PORT}/${WEBSOCKET_PORT}/g" ${CONF_DIR}/${FILENAME}
|
||||
UDP_PORT=$(($UDP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
|
||||
sed -i "s/\${UDP_PORT}/${UDP_PORT}/g" ${CONF_DIR}/${FILENAME}
|
||||
done
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [[ `uname` == "Darwin" ]]; then
|
||||
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
|
||||
else
|
||||
THIS_SCRIPT=`readlink -f $0`
|
||||
fi
|
||||
|
||||
THIS_DIR="${THIS_SCRIPT%/*}"
|
||||
BASE_DIR=$(cd $THIS_DIR/.. && pwd)
|
||||
BIN_DIR=${BASE_DIR}/bin
|
||||
|
||||
. ${BIN_DIR}/timely-balancer-env.sh
|
||||
|
||||
if [ "$NUM_PROFILES" -eq "1" ]; then
|
||||
echo ${BIN_DIR}/timely-balancer.sh
|
||||
${BIN_DIR}/timely-balancer.sh
|
||||
else
|
||||
for i in $(seq 1 $NUM_PROFILES); do
|
||||
echo ${BIN_DIR}/timely-balancer.sh $i
|
||||
${BIN_DIR}/timely-balancer.sh $i
|
||||
done
|
||||
fi
|
||||
@@ -1,3 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
pkill -9 -f 'timely-balancer-.*-exec.jar'
|
||||
File diff suppressed because it is too large.
Load diff
@@ -1,11 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
export TRUSTSTORE_FILE=
|
||||
export TRUSTSTORE_TYPE="jks";
|
||||
export TRUSTSTORE_PASS=
|
||||
export KEYSTORE_FILE=
|
||||
export KEYSTORE_TYPE="pkcs12";
|
||||
export KEYSTORE_PASS=
|
||||
|
||||
export NUM_QUERY_THREADS=4
|
||||
export NUM_INGEST_THREADS=4
|
||||
@@ -1,21 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [[ `uname` == "Darwin" ]]; then
|
||||
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
|
||||
else
|
||||
THIS_SCRIPT=`readlink -f $0`
|
||||
fi
|
||||
|
||||
THIS_DIR="${THIS_SCRIPT%/*}"
|
||||
BASE_DIR=${THIS_DIR}/../..
|
||||
TMP_DIR="${BASE_DIR}/tmp"
|
||||
CONF_DIR="${BASE_DIR}/conf"
|
||||
LIB_DIR="${BASE_DIR}/lib"
|
||||
|
||||
. ${THIS_DIR}/load-test-env.sh
|
||||
|
||||
export CLASSPATH="${CONF_DIR}:${THIS_DIR}:${LIB_DIR}/*"
|
||||
JVM_ARGS="-Xmx1G -Xms1G"
|
||||
|
||||
echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricPutLoadTest $*"
|
||||
$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricPutLoadTest $*
|
||||
@@ -1,21 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [[ `uname` == "Darwin" ]]; then
|
||||
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
|
||||
else
|
||||
THIS_SCRIPT=`readlink -f $0`
|
||||
fi
|
||||
|
||||
THIS_DIR="${THIS_SCRIPT%/*}"
|
||||
BASE_DIR=${THIS_DIR}/../..
|
||||
TMP_DIR="${BASE_DIR}/tmp"
|
||||
CONF_DIR="${BASE_DIR}/conf"
|
||||
LIB_DIR="${BASE_DIR}/lib"
|
||||
|
||||
. ${THIS_DIR}/load-test-env.sh
|
||||
|
||||
export CLASSPATH="${CONF_DIR}:${THIS_DIR}:${LIB_DIR}/*"
|
||||
JVM_ARGS="-Xmx1G -Xms1G"
|
||||
|
||||
echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricQueryLoadTest $*"
|
||||
$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricQueryLoadTest $*
|
||||
@@ -1,11 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [[ `uname` == "Darwin" ]]; then
|
||||
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
|
||||
else
|
||||
THIS_SCRIPT=`readlink -f $0`
|
||||
fi
|
||||
|
||||
THIS_DIR="${THIS_SCRIPT%/*}"
|
||||
|
||||
sed -i -- "s/#.*#/#$1#/g" $THIS_DIR/load-test-data.txt
|
||||
@@ -1,27 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
NUM_PROFILES=1
|
||||
ACCUMULO_ZOOKEEPERS=localhost:2181
|
||||
TCP_PORT_START=4242
|
||||
HTTP_PORT_START=4243
|
||||
WEBSOCKET_PORT_START=4244
|
||||
UDP_PORT_START=4245
|
||||
PORT_INCREMENT=100
|
||||
|
||||
HOST_IP=0.0.0.0
|
||||
HOST=$(hostname -s)
|
||||
HOST_FQDN=$(hostname -f)
|
||||
|
||||
GENERATE_KEYS=true
|
||||
TRUST_STORE_TYPE=
|
||||
TRUST_STORE_FILE=
|
||||
TRUST_STORE_PASSWORD=
|
||||
KEY_STORE_TYPE=
|
||||
KEY_STORE_FILE=
|
||||
KEY_STORE_PASSWORD=
|
||||
AUTHORIZATION_URL=
|
||||
AUTHORIZATION_CERTIFICATE_TYPE=
|
||||
AUTHORIZATION_CERTIFICATE_FILE=
|
||||
|
||||
REQUIRED_ROLES=
|
||||
REQUIRED_AUTHS=
|
||||
@@ -1,69 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [[ `uname` == "Darwin" ]]; then
|
||||
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
|
||||
TCNATIVE_SUFFIX="jnilib"
|
||||
else
|
||||
THIS_SCRIPT=`readlink -f $0`
|
||||
TCNATIVE_SUFFIX="so"
|
||||
fi
|
||||
|
||||
THIS_DIR="${THIS_SCRIPT%/*}"
|
||||
|
||||
set -a
|
||||
. ${THIS_DIR}/timely-balancer-env.sh
|
||||
set +a
|
||||
|
||||
if [ -n "$1" ]; then
|
||||
export INSTANCE="$1"
|
||||
PROFILE_ARG="--spring.profiles.active=${INSTANCE}"
|
||||
PID=`jps -m | grep -E 'timely-balancer-.*-exec' | grep "spring.profiles.active=${INSTANCE}" | awk '{print $1}'`
|
||||
else
|
||||
export INSTANCE="1"
|
||||
PID=`jps -m | grep -E 'timely-balancer-.*-exec' | awk '{print $1}'`
|
||||
fi
|
||||
|
||||
if [ "$PID" == "" ]; then
|
||||
|
||||
NATIVE_DIR="${THIS_DIR}/META-INF/native"
|
||||
BASE_DIR=$(cd $THIS_DIR/.. && pwd)
|
||||
TMP_DIR="${BASE_DIR}/tmp"
|
||||
export CONF_DIR="${BASE_DIR}/conf"
|
||||
LIB_DIR="${BASE_DIR}/lib"
|
||||
export LOG_DIR="${BASE_DIR}/logs"
|
||||
NUM_SERVER_THREADS=4
|
||||
|
||||
if [[ -e ${TMP_DIR} ]]; then
|
||||
rm -rf ${TMP_DIR}
|
||||
fi
|
||||
mkdir ${TMP_DIR}
|
||||
|
||||
if [[ -e ${NATIVE_DIR} ]]; then
|
||||
rm -rf ${NATIVE_DIR}
|
||||
fi
|
||||
mkdir -p ${NATIVE_DIR}
|
||||
|
||||
if [[ ! -e ${LOG_DIR} ]]; then
|
||||
mkdir ${LOG_DIR}
|
||||
fi
|
||||
|
||||
pushd ${BASE_DIR}/bin
|
||||
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
|
||||
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
|
||||
popd
|
||||
|
||||
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
|
||||
JVM_ARGS="-Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2.yml"
|
||||
JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
|
||||
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
|
||||
|
||||
JVM_ARGS="${JVM_ARGS} -Xmx2G -Xms2G -XX:NewSize=1G -XX:MaxNewSize=1G"
|
||||
JVM_ARGS="${JVM_ARGS} -XX:+UseCompressedOops -XX:+UseNUMA -Djava.net.preferIPv4Stack=true"
|
||||
JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:MaxGCPauseMillis=4000 -XX:ParallelGCThreads=20 -XX:+UseStringDeduplication"
|
||||
|
||||
echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG}"
|
||||
nohup $JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG} >> ${LOG_DIR}/timely-balancer${INSTANCE}.out 2>&1 &
|
||||
|
||||
else
|
||||
echo "timely-balancer ${INSTANCE} already running with pid ${PID}"
|
||||
fi
|
||||
@@ -1,43 +0,0 @@
|
||||
Configuration:
|
||||
status: warn
|
||||
monitorInterval: 60
|
||||
|
||||
Properties:
|
||||
Property:
|
||||
- name: logDir
|
||||
value: ${env:LOG_DIR}
|
||||
- name: PID
|
||||
value: "????"
|
||||
- name: LOG_PATTERN
|
||||
value: "%clr{%d{yyyy-MM-dd HH:mm:ss.SSS}}{faint} %clr{%5p} %clr{${sys:PID}}{magenta} %clr{---}{faint} %clr{[%15.15t]}{faint} %clr{%-40.40c{1.}}{cyan} %clr{:}{faint} %m%n%wEx"
|
||||
|
||||
Appenders:
|
||||
Console:
|
||||
name: Console
|
||||
target: SYSTEM_OUT
|
||||
follow: true
|
||||
PatternLayout:
|
||||
pattern: "${LOG_PATTERN}"
|
||||
|
||||
RollingFile:
|
||||
- name: File
|
||||
fileName: "${sys:logDir}/timely-balancer${env:INSTANCE}.log"
|
||||
filePattern: "${sys:logDir}/timely-balancer${env:INSTANCE}.log.%d{yyyy-MM-dd}-%i.gz"
|
||||
append: true
|
||||
bufferedIO: true
|
||||
bufferSize: 8192
|
||||
Policies:
|
||||
TimeBasedTriggeringPolicy:
|
||||
interval: 1
|
||||
SizeBasedTriggeringPolicy:
|
||||
size: 250MB
|
||||
DefaultRolloverStrategy:
|
||||
max: 10
|
||||
PatternLayout:
|
||||
pattern: "${LOG_PATTERN}"
|
||||
|
||||
Loggers:
|
||||
Root:
|
||||
level: info
|
||||
AppenderRef:
|
||||
- ref: File
|
||||
@@ -1,9 +0,0 @@
|
||||
trust.store.type=${TRUST_STORE_TYPE}
|
||||
trust.store.file=${TRUST_STORE_FILE}
|
||||
trust.store.password=${TRUST_STORE_PASSWORD}
|
||||
key.store.type=${KEY_STORE_TYPE}
|
||||
key.store.file=${KEY_STORE_FILE}
|
||||
key.store.password=${KEY_STORE_PASSWORD}
|
||||
authorization.url=${AUTHORIZATION_URL}
|
||||
authorization.certificate.type=${AUTHORIZATION_CERTIFICATE_TYPE}
|
||||
authorization.certificate.file=${AUTHORIZATION_CERTIFICATE_FILE}
|
||||
@@ -1,142 +0,0 @@
|
||||
timely:
|
||||
zookeeper:
|
||||
servers: ${ACCUMULO_ZOOKEEPERS}
|
||||
timeout: 30s
|
||||
security:
|
||||
session-max-age: 86400
|
||||
cache-expiration-minutes: 60
|
||||
cache-refresh-minutes: 5
|
||||
allow-anonymous-http-access: true
|
||||
allow-anonymous-ws-access: true
|
||||
serverSsl:
|
||||
key-store-type: ${KEY_STORE_TYPE}
|
||||
key-store-file: ${KEY_STORE_FILE}
|
||||
key-store-password: ${KEY_STORE_PASSWORD}
|
||||
use-generated-keypair: ${GENERATE_KEYS}
|
||||
ignore-ssl-handshake-errors: true
|
||||
trust-store-type: ${TRUST_STORE_TYPE}
|
||||
trust-store-file: ${TRUST_STORE_FILE}
|
||||
trust-store-password: ${TRUST_STORE_PASSWORD}
|
||||
clientSsl:
|
||||
key-store-type: ${KEY_STORE_TYPE}
|
||||
key-store-file: ${KEY_STORE_FILE}
|
||||
key-store-password: ${KEY_STORE_PASSWORD}
|
||||
trust-store-type: ${TRUST_STORE_TYPE}
|
||||
trust-store-file: ${TRUST_STORE_FILE}
|
||||
trust-store-password: ${TRUST_STORE_PASSWORD}
|
||||
jwtSsl:
|
||||
key-store-type: ${AUTHORIZATION_CERTIFICATE_TYPE}
|
||||
key-store-file: ${AUTHORIZATION_CERTIFICATE_FILE}
|
||||
key-store-password: ${AUTHORIZATION_CERTIFICATE_PASSWORD}
|
||||
cache:
|
||||
enabled: false
|
||||
metric-age-off-hours:
|
||||
default: 26
|
||||
nonCachedMetrics:
|
||||
balancer:
|
||||
checkServerHealthInterval: 10000
|
||||
serverFailuresBeforeDown: 6
|
||||
serverSuccessesBeforeUp: 1
|
||||
assignmentFile: /timely/balancer/assignments.csv
|
||||
server:
|
||||
ip: ${HOST_IP}
|
||||
tcpPort: ${TCP_PORT}
|
||||
numTcpPools: 1
|
||||
tcpBufferSize: 100
|
||||
tcpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
udpPort: ${UDP_PORT}
|
||||
udpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
http:
|
||||
host: ${HOST}
|
||||
ip: ${HOST_IP}
|
||||
port: ${HTTP_PORT}
|
||||
redirect-path: /secure-me
|
||||
strict-transport-max-age: 604800
|
||||
cors:
|
||||
allow-any-origin: false
|
||||
allow-null-origin: false
|
||||
allow-credentials: true
|
||||
allowed-origins:
|
||||
allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
|
||||
allowed-headers: content-type
|
||||
httpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
websocket:
|
||||
ip: ${HOST_IP}
|
||||
port: ${WEBSOCKET_PORT}
|
||||
timeout: 60
|
||||
subscription-lag: 120
|
||||
scanner-batch-size: 10000
|
||||
flush-interval-seconds: 30
|
||||
scanner-read-ahead: 1
|
||||
subscription-batch-size: 5000
|
||||
incomingBufferSize: 50000000
|
||||
wsClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
|
||||
logging:
|
||||
level:
|
||||
ROOT: info
|
||||
timey: debug
|
||||
@@ -1,142 +0,0 @@
|
||||
timely:
|
||||
zookeeper:
|
||||
servers: ${ACCUMULO_ZOOKEEPERS}
|
||||
timeout: 30s
|
||||
security:
|
||||
session-max-age: 86400
|
||||
cache-expiration-minutes: 60
|
||||
cache-refresh-minutes: 5
|
||||
allow-anonymous-http-access: true
|
||||
allow-anonymous-ws-access: true
|
||||
serverSsl:
|
||||
key-store-type:
|
||||
key-store-file:
|
||||
key-store-password:
|
||||
use-generated-keypair: true
|
||||
ignore-ssl-handshake-errors: true
|
||||
trust-store-type:
|
||||
trust-store-file:
|
||||
trust-store-password:
|
||||
clientSsl:
|
||||
key-store-type:
|
||||
key-store-file:
|
||||
key-store-password:
|
||||
trust-store-type:
|
||||
trust-store-file:
|
||||
trust-store-password:
|
||||
jwtSsl:
|
||||
key-store-type:
|
||||
key-store-file:
|
||||
key-store-password:
|
||||
cache:
|
||||
enabled: false
|
||||
metric-age-off-hours:
|
||||
default: 26
|
||||
nonCachedMetrics:
|
||||
balancer:
|
||||
checkServerHealthInterval: 10000
|
||||
serverFailuresBeforeDown: 6
|
||||
serverSuccessesBeforeUp: 1
|
||||
assignmentFile: /timely/balancer/assignments.csv
|
||||
server:
|
||||
ip: 0.0.0.0
|
||||
tcpPort: 4242
|
||||
numTcpPools: 1
|
||||
tcpBufferSize: 100
|
||||
tcpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
udpPort: 4245
|
||||
udpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
http:
|
||||
host: localhost
|
||||
ip: 0.0.0.0
|
||||
port: 4243
|
||||
redirect-path: /secure-me
|
||||
strict-transport-max-age: 604800
|
||||
cors:
|
||||
allow-any-origin: false
|
||||
allow-null-origin: false
|
||||
allow-credentials: true
|
||||
allowed-origins:
|
||||
allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
|
||||
allowed-headers: content-type
|
||||
httpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
websocket:
|
||||
ip: 0.0.0.0
|
||||
port: 5244
|
||||
timeout: 60
|
||||
subscription-lag: 120
|
||||
scanner-batch-size: 10000
|
||||
flush-interval-seconds: 30
|
||||
scanner-read-ahead: 1
|
||||
subscription-batch-size: 5000
|
||||
incomingBufferSize: 50000000
|
||||
wsClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
|
||||
logging:
|
||||
level:
|
||||
ROOT: info
|
||||
timey: debug
|
||||
@@ -1,338 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely</artifactId>
|
||||
<version>2.1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>timely-balancer</artifactId>
|
||||
<description>Timely Balancer</description>
|
||||
<properties>
|
||||
<start-class>timely.balancer.TimelyBalancer</start-class>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-annotations</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.datatype</groupId>
|
||||
<artifactId>jackson-datatype-jdk8</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.github.ben-manes.caffeine</groupId>
|
||||
<artifactId>caffeine</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.google.flatbuffers</groupId>
|
||||
<artifactId>flatbuffers-java</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>commons-beanutils</groupId>
|
||||
<artifactId>commons-beanutils</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>commons-logging</groupId>
|
||||
<artifactId>commons-logging</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.github.classgraph</groupId>
|
||||
<artifactId>classgraph</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.netty</groupId>
|
||||
<artifactId>netty-all</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.sourceforge.javacsv</groupId>
|
||||
<artifactId>javacsv</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.accumulo</groupId>
|
||||
<artifactId>accumulo-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.accumulo</groupId>
|
||||
<artifactId>accumulo-minicluster</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-collections4</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-lang3</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-pool2</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.curator</groupId>
|
||||
<artifactId>curator-client</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.curator</groupId>
|
||||
<artifactId>curator-framework</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.curator</groupId>
|
||||
<artifactId>curator-recipes</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.curator</groupId>
|
||||
<artifactId>curator-x-discovery</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.hadoop</groupId>
|
||||
<artifactId>hadoop-client</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.hadoop</groupId>
|
||||
<artifactId>hadoop-common</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.hadoop</groupId>
|
||||
<artifactId>hadoop-minicluster</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-1.2-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-webflux</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-websocket</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter</artifactId>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-logging</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-log4j2</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-validation</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-context</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-bootstrap</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.security</groupId>
|
||||
<artifactId>spring-security-config</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.security</groupId>
|
||||
<artifactId>spring-security-web</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-client</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-common</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-server</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.lmax</groupId>
|
||||
<artifactId>disruptor</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.netty</groupId>
|
||||
<artifactId>netty-tcnative-boringssl-static</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-api</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-core</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.javassist</groupId>
|
||||
<artifactId>javassist</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.jsoup</groupId>
|
||||
<artifactId>jsoup</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-test-autoconfigure</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-assembly-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>assemble</id>
|
||||
<goals>
|
||||
<goal>single</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<descriptors>
|
||||
<descriptor>src/main/dist/assembly.xml</descriptor>
|
||||
</descriptors>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-dependency-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>copy-dependencies</id>
|
||||
<goals>
|
||||
<goal>copy-dependencies</goal>
|
||||
</goals>
|
||||
<phase>prepare-package</phase>
|
||||
<configuration>
|
||||
<excludeArtifactIds>junit</excludeArtifactIds>
|
||||
<outputDirectory>${project.build.directory}/lib</outputDirectory>
|
||||
</configuration>
|
||||
</execution>
|
||||
<execution>
|
||||
<id>unpack-native</id>
|
||||
<goals>
|
||||
<goal>unpack</goal>
|
||||
</goals>
|
||||
<phase>process-test-classes</phase>
|
||||
<configuration>
|
||||
<artifactItems>
|
||||
<artifactItem>
|
||||
<groupId>io.netty</groupId>
|
||||
<artifactId>netty-tcnative-boringssl-static</artifactId>
|
||||
<version>${version.netty-tcnative}</version>
|
||||
<type>jar</type>
|
||||
<overWrite>false</overWrite>
|
||||
<outputDirectory>${project.build.directory}/native</outputDirectory>
|
||||
<includes>**/*.so</includes>
|
||||
</artifactItem>
|
||||
</artifactItems>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-failsafe-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-jar-plugin</artifactId>
|
||||
<version>3.1.2</version>
|
||||
<configuration>
|
||||
<archive>
|
||||
<manifest>
|
||||
<!--suppress UnresolvedMavenProperty, MavenModelInspection -->
|
||||
<mainClass>${start-class}</mainClass>
|
||||
<addDefaultImplementationEntries>true</addDefaultImplementationEntries>
|
||||
</manifest>
|
||||
</archive>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>build-helper-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>add-source</id>
|
||||
<goals>
|
||||
<goal>add-source</goal>
|
||||
</goals>
|
||||
<phase>generate-sources</phase>
|
||||
<configuration>
|
||||
<sources>
|
||||
<source>src/main/generated-java</source>
|
||||
</sources>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
<extensions>
|
||||
<!-- Use os-maven-plugin to initialize the "os.detected" properties -->
|
||||
<extension>
|
||||
<groupId>kr.motd.maven</groupId>
|
||||
<artifactId>os-maven-plugin</artifactId>
|
||||
</extension>
|
||||
</extensions>
|
||||
</build>
|
||||
<profiles>
|
||||
<profile>
|
||||
<id>docker</id>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>com.spotify</groupId>
|
||||
<artifactId>docker-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</profile>
|
||||
</profiles>
|
||||
</project>
|
||||
Vendored
-55
@@ -1,55 +0,0 @@
|
||||
<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.3"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.3 http://maven.apache.org/xsd/assembly-1.1.3.xsd">
|
||||
<id>dist</id>
|
||||
<formats>
|
||||
<format>tar.gz</format>
|
||||
</formats>
|
||||
<fileSets>
|
||||
<fileSet>
|
||||
<directory>${project.basedir}</directory>
|
||||
<outputDirectory></outputDirectory>
|
||||
<includes>
|
||||
<include>README*</include>
|
||||
<include>LICENSE*</include>
|
||||
<include>NOTICE*</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>${project.build.directory}/lib</directory>
|
||||
<outputDirectory>lib</outputDirectory>
|
||||
<includes>
|
||||
<include>netty-tcnative-boringssl-static*.jar</include>
|
||||
<include>netty-all*.jar</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>${project.build.directory}</directory>
|
||||
<outputDirectory>bin</outputDirectory>
|
||||
<includes>
|
||||
<include>timely-balancer-*-exec.jar</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>${project.basedir}/bin</directory>
|
||||
<outputDirectory>bin</outputDirectory>
|
||||
<includes>
|
||||
<include>**/*</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>${project.basedir}/conf</directory>
|
||||
<outputDirectory>conf</outputDirectory>
|
||||
<includes>
|
||||
<include>*</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>${project.basedir}/src/main/resources/webapp</directory>
|
||||
<outputDirectory>bin/webapp</outputDirectory>
|
||||
<includes>
|
||||
<include>*</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
</fileSets>
|
||||
</assembly>
|
||||
@@ -1,11 +0,0 @@
|
||||
FROM azul/zulu-openjdk-alpine:11
|
||||
|
||||
LABEL version=${project.version} \
|
||||
run="docker run ${docker.image.prefix}${project.artifactId}:latest" \
|
||||
description="${project.description}"
|
||||
|
||||
ADD ${project.build.finalName}-exec.jar /app.jar
|
||||
RUN apk add libc6-compat curl
|
||||
|
||||
EXPOSE 54322 54323 54324 54325
|
||||
ENTRYPOINT ["java","-jar","app.jar"]
|
||||
@@ -1,121 +0,0 @@
|
||||
package timely.balancer;
|
||||
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import com.google.common.util.concurrent.AtomicDouble;
|
||||
|
||||
public class ArrivalRate {
|
||||
|
||||
private AtomicLong shortTermArrivals = new AtomicLong(0);
|
||||
private AtomicDouble shortTermRate = new AtomicDouble(0);
|
||||
private AtomicLong shortTermLastReset = new AtomicLong(0);
|
||||
|
||||
private AtomicLong mediumTermArrivals = new AtomicLong(0);
|
||||
private AtomicDouble mediumTermRate = new AtomicDouble(0);
|
||||
private AtomicLong mediumTermLastReset = new AtomicLong(0);
|
||||
|
||||
private AtomicLong longTermArrivals = new AtomicLong(0);
|
||||
private AtomicDouble longTermRate = new AtomicDouble(0);
|
||||
private AtomicLong longTermLastReset = new AtomicLong(0);
|
||||
|
||||
public ArrivalRate(ScheduledExecutorService arrivalRateExecutor) {
|
||||
long now = System.currentTimeMillis();
|
||||
shortTermLastReset.set(now);
|
||||
mediumTermLastReset.set(now);
|
||||
longTermLastReset.set(now);
|
||||
|
||||
// 2 minute
|
||||
arrivalRateExecutor.scheduleAtFixedRate(() -> {
|
||||
resetShort();
|
||||
}, 2, 2, TimeUnit.MINUTES);
|
||||
|
||||
// 5 minute
|
||||
arrivalRateExecutor.scheduleAtFixedRate(() -> {
|
||||
resetMedium();
|
||||
}, 5, 5, TimeUnit.MINUTES);
|
||||
|
||||
// 15 minute
|
||||
arrivalRateExecutor.scheduleAtFixedRate(() -> {
|
||||
resetLong();
|
||||
}, 15, 15, TimeUnit.MINUTES);
|
||||
}
|
||||
|
||||
private void resetShort() {
|
||||
long now = System.currentTimeMillis();
|
||||
long lastReset = shortTermLastReset.get();
|
||||
if (now == lastReset) {
|
||||
shortTermRate.set(0);
|
||||
} else {
|
||||
shortTermRate.set(shortTermArrivals.get() / (((double) (now - lastReset)) / 1000));
|
||||
}
|
||||
shortTermArrivals.set(0);
|
||||
shortTermLastReset.set(now);
|
||||
}
|
||||
|
||||
private void resetMedium() {
|
||||
long now = System.currentTimeMillis();
|
||||
long lastReset = mediumTermLastReset.get();
|
||||
if (now == lastReset) {
|
||||
mediumTermRate.set(0);
|
||||
} else {
|
||||
mediumTermRate.set(mediumTermArrivals.get() / (((double) (now - lastReset)) / 1000));
|
||||
}
|
||||
mediumTermArrivals.set(0);
|
||||
mediumTermLastReset.set(now);
|
||||
}
|
||||
|
||||
private void resetLong() {
|
||||
long now = System.currentTimeMillis();
|
||||
long lastReset = longTermLastReset.get();
|
||||
if (now == lastReset) {
|
||||
longTermRate.set(0);
|
||||
} else {
|
||||
longTermRate.set(longTermArrivals.get() / (((double) (now - lastReset)) / 1000));
|
||||
}
|
||||
longTermArrivals.set(0);
|
||||
longTermLastReset.set(now);
|
||||
}
|
||||
|
||||
public void arrived() {
|
||||
shortTermArrivals.incrementAndGet();
|
||||
mediumTermArrivals.incrementAndGet();
|
||||
longTermArrivals.incrementAndGet();
|
||||
}
|
||||
|
||||
private double getShortRate(long now) {
|
||||
double rate = shortTermRate.get();
|
||||
long lastReset = shortTermLastReset.get();
|
||||
if (rate > 0 || now == lastReset) {
|
||||
return rate;
|
||||
} else {
|
||||
return shortTermArrivals.get() / (((double) (now - lastReset)) / 1000);
|
||||
}
|
||||
}
|
||||
|
||||
private double getMediumRate(long now) {
|
||||
double rate = mediumTermRate.get();
|
||||
long lastReset = mediumTermLastReset.get();
|
||||
if (rate > 0 || now == lastReset) {
|
||||
return rate;
|
||||
} else {
|
||||
return mediumTermArrivals.get() / (((double) (now - lastReset)) / 1000);
|
||||
}
|
||||
}
|
||||
|
||||
private double getLongRate(long now) {
|
||||
double rate = longTermRate.get();
|
||||
long lastReset = longTermLastReset.get();
|
||||
if (rate > 0 || now == lastReset) {
|
||||
return rate;
|
||||
} else {
|
||||
return longTermArrivals.get() / (((double) (now - lastReset)) / 1000);
|
||||
}
|
||||
}
|
||||
|
||||
public double getRate() {
|
||||
long now = System.currentTimeMillis();
|
||||
return 0.10 * getLongRate(now) + 0.10 * getMediumRate(now) + 0.80 * getShortRate(now);
|
||||
}
|
||||
}
|
||||
@@ -1,14 +0,0 @@
|
||||
package timely.balancer;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.autoconfigure.web.servlet.error.ErrorMvcAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
|
||||
@SpringBootApplication(scanBasePackages = {"timely.balancer", "timely.common"}, exclude = {ErrorMvcAutoConfiguration.class})
|
||||
public class TimelyBalancer {
|
||||
|
||||
public static void main(String[] args) {
|
||||
new SpringApplicationBuilder(TimelyBalancer.class).web(WebApplicationType.NONE).run(args);
|
||||
}
|
||||
}
|
||||
@@ -1,526 +0,0 @@
|
||||
package timely.balancer.component;
|
||||
|
||||
import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
|
||||
import static timely.Constants.NON_CACHED_METRICS;
|
||||
import static timely.Constants.NON_CACHED_METRICS_LOCK_PATH;
|
||||
import static timely.Constants.SERVICE_DISCOVERY_PATH;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import org.apache.curator.RetryPolicy;
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.CuratorFrameworkFactory;
|
||||
import org.apache.curator.retry.RetryForever;
|
||||
import org.apache.hadoop.util.StringUtils;
|
||||
import org.apache.zookeeper.CreateMode;
|
||||
import org.apache.zookeeper.data.Stat;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import io.netty.bootstrap.AbstractBootstrap;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.bootstrap.ServerBootstrap;
|
||||
import io.netty.buffer.PooledByteBufAllocator;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelOption;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.epoll.EpollChannelOption;
|
||||
import io.netty.channel.epoll.EpollDatagramChannel;
|
||||
import io.netty.channel.epoll.EpollEventLoopGroup;
|
||||
import io.netty.channel.epoll.EpollServerSocketChannel;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.DatagramChannel;
|
||||
import io.netty.channel.socket.ServerSocketChannel;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioDatagramChannel;
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
|
||||
import io.netty.handler.codec.Delimiters;
|
||||
import io.netty.handler.codec.http.HttpContentCompressor;
|
||||
import io.netty.handler.codec.http.HttpContentDecompressor;
|
||||
import io.netty.handler.codec.http.HttpObjectAggregator;
|
||||
import io.netty.handler.codec.http.HttpRequestDecoder;
|
||||
import io.netty.handler.codec.http.HttpResponseEncoder;
|
||||
import io.netty.handler.codec.http.HttpServerCodec;
|
||||
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
import io.netty.handler.ssl.ClientAuth;
|
||||
import io.netty.handler.ssl.OpenSslServerContext;
|
||||
import io.netty.handler.ssl.OpenSslServerSessionContext;
|
||||
import io.netty.handler.ssl.SslContext;
|
||||
import io.netty.handler.ssl.SslContextBuilder;
|
||||
import io.netty.handler.ssl.util.SelfSignedCertificate;
|
||||
import io.netty.handler.stream.ChunkedWriteHandler;
|
||||
import io.netty.handler.timeout.IdleStateHandler;
|
||||
import io.netty.util.NettyRuntime;
|
||||
import io.netty.util.internal.SystemPropertyUtil;
|
||||
import timely.auth.JWTTokenHandler;
|
||||
import timely.balancer.configuration.BalancerHttpProperties;
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.configuration.BalancerServerProperties;
|
||||
import timely.balancer.configuration.BalancerWebsocketProperties;
|
||||
import timely.balancer.connection.http.HttpClientPool;
|
||||
import timely.balancer.connection.tcp.TcpClientPool;
|
||||
import timely.balancer.connection.udp.UdpClientPool;
|
||||
import timely.balancer.connection.ws.WsClientPool;
|
||||
import timely.balancer.netty.http.HttpRelayHandler;
|
||||
import timely.balancer.netty.tcp.TcpRelayHandler;
|
||||
import timely.balancer.netty.udp.UdpRelayHandler;
|
||||
import timely.balancer.netty.ws.WsRelayHandler;
|
||||
import timely.client.http.HttpClient;
|
||||
import timely.common.component.AuthenticationService;
|
||||
import timely.common.configuration.SecurityProperties;
|
||||
import timely.common.configuration.SslClientProperties;
|
||||
import timely.common.configuration.SslHelper;
|
||||
import timely.common.configuration.SslServerProperties;
|
||||
import timely.common.configuration.ZookeeperProperties;
|
||||
import timely.netty.http.HttpStaticFileServerHandler;
|
||||
import timely.netty.http.NonSslRedirectHandler;
|
||||
import timely.netty.http.TimelyExceptionHandler;
|
||||
import timely.netty.http.auth.X509LoginRequestHandler;
|
||||
import timely.netty.tcp.MetricsBufferDecoder;
|
||||
import timely.netty.tcp.TcpDecoder;
|
||||
import timely.netty.tcp.TcpVersionHandler;
|
||||
import timely.netty.udp.UdpDecoder;
|
||||
import timely.netty.udp.UdpPacketToByteBuf;
|
||||
import timely.netty.websocket.WebSocketFullRequestHandler;
|
||||
import timely.netty.websocket.WebSocketRequestDecoder;
|
||||
import timely.netty.websocket.subscription.WSTimelyExceptionHandler;
|
||||
|
||||
@Component
|
||||
@ConditionalOnMissingBean(Balancer.class)
|
||||
public class Balancer {
|
||||
|
||||
final public static String LEADER_LATCH_PATH = "/timely/balancer/leader";
|
||||
final public static String ASSIGNMENTS_LOCK_PATH = "/timely/balancer/assignments/lock";
|
||||
final public static String ASSIGNMENTS_LAST_UPDATED_PATH = "/timely/balancer/assignments/lastUpdated";
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(Balancer.class);
|
||||
private static final String EPOLL_MIN_VERSION = "2.7";
|
||||
private static final String OS_NAME = "os.name";
|
||||
private static final String OS_VERSION = "os.version";
|
||||
private static final String WS_PATH = "/websocket";
|
||||
protected static final CountDownLatch LATCH = new CountDownLatch(1);
|
||||
|
||||
private BalancerProperties balancerProperties;
|
||||
private SecurityProperties securityProperties;
|
||||
private SslServerProperties sslServerProperties;
|
||||
private SslClientProperties sslClientProperties;
|
||||
private ZookeeperProperties zookeeperProperties;
|
||||
private BalancerServerProperties balancerServerProperties;
|
||||
private BalancerHttpProperties balancerHttpProperties;
|
||||
private BalancerWebsocketProperties balancerWebsocketProperties;
|
||||
private ApplicationContext applicationContext;
|
||||
private AuthenticationService authenticationService;
|
||||
private CuratorFramework curatorFramework;
|
||||
private int quietPeriod;
|
||||
|
||||
private EventLoopGroup tcpWorkerGroup = null;
|
||||
private EventLoopGroup tcpBossGroup = null;
|
||||
private EventLoopGroup httpWorkerGroup = null;
|
||||
private EventLoopGroup httpBossGroup = null;
|
||||
private EventLoopGroup wsWorkerGroup = null;
|
||||
private EventLoopGroup wsBossGroup = null;
|
||||
private EventLoopGroup udpBossGroup = null;
|
||||
private EventLoopGroup udpWorkerGroup = null;
|
||||
private MetricResolver metricResolver;
|
||||
protected Channel tcpChannelHandle = null;
|
||||
protected Channel httpChannelHandle = null;
|
||||
protected Channel wsChannelHandle = null;
|
||||
protected List<Channel> udpChannelHandleList = new ArrayList<>();
|
||||
protected int DEFAULT_EVENT_LOOP_THREADS;
|
||||
|
||||
private String[] zkPaths = new String[] {LEADER_LATCH_PATH, ASSIGNMENTS_LAST_UPDATED_PATH, ASSIGNMENTS_LOCK_PATH, SERVICE_DISCOVERY_PATH,
|
||||
NON_CACHED_METRICS, NON_CACHED_METRICS_LOCK_PATH};
|
||||
|
||||
private static boolean useEpoll() {
|
||||
final String osName = SystemPropertyUtil.get(OS_NAME).toLowerCase().trim();
|
||||
final String osVersion = SystemPropertyUtil.get(OS_VERSION).toLowerCase();
|
||||
// split at periods, keep the first two, join with a period
|
||||
final String majMinVers = StringUtils.join(".", Arrays.stream(osVersion.split("\\.")).limit(2).collect(Collectors.toList()));
|
||||
if (osName.startsWith("linux")) {
|
||||
Runtime.Version currentVersion = Runtime.Version.parse(majMinVers);
|
||||
Runtime.Version epollMinVersion = Runtime.Version.parse(EPOLL_MIN_VERSION);
|
||||
return currentVersion.compareTo(epollMinVersion) > 0;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public Balancer(ApplicationContext applicationContext, AuthenticationService authenticationService, MetricResolver metricResolver,
|
||||
BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
|
||||
SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
|
||||
BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
|
||||
this.quietPeriod = balancerServerProperties.getShutdownQuietPeriod();
|
||||
this.applicationContext = applicationContext;
|
||||
this.authenticationService = authenticationService;
|
||||
this.metricResolver = metricResolver;
|
||||
this.balancerProperties = balancerProperties;
|
||||
this.zookeeperProperties = zookeeperProperties;
|
||||
this.securityProperties = securityProperties;
|
||||
this.sslServerProperties = sslServerProperties;
|
||||
this.sslClientProperties = sslClientProperties;
|
||||
this.balancerServerProperties = balancerServerProperties;
|
||||
this.balancerHttpProperties = balancerHttpProperties;
|
||||
this.balancerWebsocketProperties = balancerWebsocketProperties;
|
||||
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
public void startup() {
|
||||
log.info("Starting {}", this.getClass().getSimpleName());
|
||||
try {
|
||||
// start curator framework
|
||||
RetryPolicy retryPolicy = new RetryForever(1000);
|
||||
int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
|
||||
curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
|
||||
curatorFramework.start();
|
||||
ensureZkPaths(curatorFramework, zkPaths);
|
||||
|
||||
JWTTokenHandler.init(securityProperties, null);
|
||||
final boolean useEpoll = useEpoll();
|
||||
Class<? extends ServerSocketChannel> channelClass;
|
||||
Class<? extends Channel> datagramChannelClass;
|
||||
if (useEpoll) {
|
||||
tcpWorkerGroup = new EpollEventLoopGroup();
|
||||
tcpBossGroup = new EpollEventLoopGroup();
|
||||
httpWorkerGroup = new EpollEventLoopGroup();
|
||||
httpBossGroup = new EpollEventLoopGroup();
|
||||
wsWorkerGroup = new EpollEventLoopGroup();
|
||||
wsBossGroup = new EpollEventLoopGroup();
|
||||
udpWorkerGroup = new EpollEventLoopGroup();
|
||||
udpBossGroup = new EpollEventLoopGroup();
|
||||
channelClass = EpollServerSocketChannel.class;
|
||||
datagramChannelClass = EpollDatagramChannel.class;
|
||||
} else {
|
||||
tcpWorkerGroup = new NioEventLoopGroup();
|
||||
tcpBossGroup = new NioEventLoopGroup();
|
||||
httpWorkerGroup = new NioEventLoopGroup();
|
||||
httpBossGroup = new NioEventLoopGroup();
|
||||
wsWorkerGroup = new NioEventLoopGroup();
|
||||
wsBossGroup = new NioEventLoopGroup();
|
||||
udpWorkerGroup = new NioEventLoopGroup();
|
||||
udpBossGroup = new NioEventLoopGroup();
|
||||
channelClass = NioServerSocketChannel.class;
|
||||
datagramChannelClass = NioDatagramChannel.class;
|
||||
}
|
||||
log.info("Using channel class {}", channelClass.getSimpleName());
|
||||
|
||||
List<TcpClientPool> tcpClientPools = new ArrayList<>();
|
||||
for (int x = 0; x < balancerServerProperties.getNumTcpPools(); x++) {
|
||||
tcpClientPools.add(new TcpClientPool(balancerServerProperties));
|
||||
}
|
||||
|
||||
final ServerBootstrap tcpServer = new ServerBootstrap();
|
||||
tcpServer.group(tcpBossGroup, tcpWorkerGroup);
|
||||
tcpServer.channel(channelClass);
|
||||
tcpServer.handler(new LoggingHandler());
|
||||
tcpServer.childHandler(setupTcpChannel(metricResolver, tcpClientPools));
|
||||
tcpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
|
||||
tcpServer.option(ChannelOption.SO_BACKLOG, 128);
|
||||
tcpServer.childOption(ChannelOption.SO_KEEPALIVE, true);
|
||||
final int tcpPort = balancerServerProperties.getTcpPort();
|
||||
final String tcpIp = balancerServerProperties.getIp();
|
||||
tcpChannelHandle = bind(tcpServer, tcpIp, tcpPort);
|
||||
final String tcpAddress = ((InetSocketAddress) tcpChannelHandle.localAddress()).getAddress().getHostAddress();
|
||||
|
||||
final int httpPort = balancerHttpProperties.getPort();
|
||||
final String httpIp = balancerHttpProperties.getIp();
|
||||
SslContext sslCtx = createSSLContext(sslServerProperties);
|
||||
if (sslCtx instanceof OpenSslServerContext) {
|
||||
OpenSslServerContext openssl = (OpenSslServerContext) sslCtx;
|
||||
String application = "Timely_" + httpPort;
|
||||
OpenSslServerSessionContext opensslCtx = openssl.sessionContext();
|
||||
opensslCtx.setSessionCacheEnabled(true);
|
||||
opensslCtx.setSessionCacheSize(128);
|
||||
opensslCtx.setSessionIdContext(application.getBytes(StandardCharsets.UTF_8));
|
||||
opensslCtx.setSessionTimeout(securityProperties.getSessionMaxAge());
|
||||
}
|
||||
SSLContext clientSSLContext = createSSLClientContext(sslClientProperties);
|
||||
final ServerBootstrap httpServer = new ServerBootstrap();
|
||||
httpServer.group(httpBossGroup, httpWorkerGroup);
|
||||
httpServer.channel(channelClass);
|
||||
httpServer.handler(new LoggingHandler());
|
||||
HttpClientPool httpClientPool = new HttpClientPool(securityProperties, sslClientProperties, balancerHttpProperties, clientSSLContext);
|
||||
httpServer.childHandler(setupHttpChannel(balancerHttpProperties, sslServerProperties, sslCtx, metricResolver, httpClientPool));
|
||||
httpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
|
||||
httpServer.option(ChannelOption.SO_BACKLOG, 128);
|
||||
httpServer.childOption(ChannelOption.SO_KEEPALIVE, true);
|
||||
httpChannelHandle = bind(httpServer, httpIp, httpPort);
|
||||
final String httpAddress = ((InetSocketAddress) httpChannelHandle.localAddress()).getAddress().getHostAddress();
|
||||
|
||||
final int wsPort = balancerWebsocketProperties.getPort();
|
||||
final String wsIp = balancerWebsocketProperties.getIp();
|
||||
final ServerBootstrap wsServer = new ServerBootstrap();
|
||||
wsServer.group(wsBossGroup, wsWorkerGroup);
|
||||
wsServer.channel(channelClass);
|
||||
wsServer.handler(new LoggingHandler());
|
||||
WsClientPool wsClientPool = new WsClientPool(balancerProperties, balancerWebsocketProperties, sslClientProperties, clientSSLContext);
|
||||
wsServer.childHandler(setupWSChannel(balancerWebsocketProperties, securityProperties, sslCtx, metricResolver, wsClientPool));
|
||||
wsServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
|
||||
wsServer.option(ChannelOption.SO_BACKLOG, 128);
|
||||
wsServer.childOption(ChannelOption.SO_KEEPALIVE, true);
|
||||
wsChannelHandle = bind(wsServer, wsIp, wsPort);
|
||||
final String wsAddress = ((InetSocketAddress) wsChannelHandle.localAddress()).getAddress().getHostAddress();
|
||||
|
||||
final int udpPort = balancerServerProperties.getUdpPort();
|
||||
final String udpIp = balancerServerProperties.getIp();
|
||||
UdpClientPool udpClientPool = new UdpClientPool(balancerServerProperties);
|
||||
for (int n = 0; n < DEFAULT_EVENT_LOOP_THREADS; n++) {
|
||||
final Bootstrap udpServer = new Bootstrap();
|
||||
udpServer.group(udpBossGroup);
|
||||
udpServer.channel(datagramChannelClass);
|
||||
udpServer.handler(setupUdpChannel(metricResolver, udpClientPool));
|
||||
udpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
|
||||
udpServer.option(EpollChannelOption.SO_REUSEADDR, true);
|
||||
udpServer.option(EpollChannelOption.SO_REUSEPORT, true);
|
||||
udpChannelHandleList.add(bind(udpServer, udpIp, udpPort));
|
||||
}
|
||||
log.info("Balancer started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, {}:{} for WebSocket traffic, and {}:{} for UDP traffic",
|
||||
tcpAddress, tcpPort, httpAddress, httpPort, wsAddress, wsPort, wsAddress, udpPort);
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage(), e);
|
||||
SpringApplication.exit(applicationContext, () -> 0);
|
||||
}
|
||||
}
|
||||
|
||||
protected Channel bind(AbstractBootstrap server, String ip, int port) {
|
||||
Channel channel = null;
|
||||
try {
|
||||
channel = server.bind(ip, port).sync().channel();
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Failed to bind to {}:{} -- {}", ip, port, e.getMessage());
|
||||
}
|
||||
if (channel == null) {
|
||||
throw new IllegalStateException("Failed to bind to port:" + ip + ":" + port);
|
||||
}
|
||||
return channel;
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
public void shutdown() {
|
||||
List<ChannelFuture> channelFutures = new ArrayList<>();
|
||||
|
||||
if (tcpChannelHandle != null) {
|
||||
log.info("Closing tcpChannelHandle");
|
||||
channelFutures.add(tcpChannelHandle.close());
|
||||
}
|
||||
|
||||
if (httpChannelHandle != null) {
|
||||
log.info("Closing httpChannelHandle");
|
||||
channelFutures.add(httpChannelHandle.close());
|
||||
}
|
||||
|
||||
if (wsChannelHandle != null) {
|
||||
log.info("Closing wsChannelHandle");
|
||||
channelFutures.add(wsChannelHandle.close());
|
||||
}
|
||||
|
||||
int udpChannel = 1;
|
||||
for (Channel c : udpChannelHandleList) {
|
||||
log.info("Closing udpChannelHandle #" + udpChannel++);
|
||||
channelFutures.add(c.close());
|
||||
}
|
||||
|
||||
// wait for the channels to shutdown
|
||||
channelFutures.forEach(f -> {
|
||||
try {
|
||||
f.get();
|
||||
} catch (final Exception e) {
|
||||
log.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
|
||||
}
|
||||
});
|
||||
|
||||
List<Future<?>> groupFutures = new ArrayList<>();
|
||||
|
||||
if (tcpBossGroup != null) {
|
||||
log.info("Shutting down tcpBossGroup");
|
||||
groupFutures.add(tcpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (tcpWorkerGroup != null) {
|
||||
log.info("Shutting down tcpWorkerGroup");
|
||||
groupFutures.add(tcpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (httpBossGroup != null) {
|
||||
log.info("Shutting down httpBossGroup");
|
||||
groupFutures.add(httpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (httpWorkerGroup != null) {
|
||||
log.info("Shutting down httpWorkerGroup");
|
||||
groupFutures.add(httpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (wsBossGroup != null) {
|
||||
log.info("Shutting down wsBossGroup");
|
||||
groupFutures.add(wsBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (wsWorkerGroup != null) {
|
||||
log.info("Shutting down wsWorkerGroup");
|
||||
groupFutures.add(wsWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (udpBossGroup != null) {
|
||||
log.info("Shutting down udpBossGroup");
|
||||
groupFutures.add(udpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
if (udpWorkerGroup != null) {
|
||||
log.info("Shutting down udpWorkerGroup");
|
||||
groupFutures.add(udpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
groupFutures.parallelStream().forEach(f -> {
|
||||
try {
|
||||
f.get();
|
||||
} catch (final Exception e) {
|
||||
log.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
|
||||
}
|
||||
});
|
||||
|
||||
try {
|
||||
log.info("Closing WebSocketRequestDecoder");
|
||||
WebSocketRequestDecoder.close();
|
||||
} catch (Exception e) {
|
||||
log.error("Error closing WebSocketRequestDecoder during shutdown", e);
|
||||
}
|
||||
log.info("{} shut down.", this.getClass().getSimpleName());
|
||||
}
|
||||
|
||||
protected SslContext createSSLContext(SslServerProperties sslServerProperties) throws Exception {
|
||||
|
||||
Boolean generate = sslServerProperties.isUseGeneratedKeypair();
|
||||
SslContextBuilder ssl;
|
||||
if (generate) {
|
||||
log.warn("Using generated self signed server certificate");
|
||||
Date begin = new Date();
|
||||
Date end = new Date(begin.getTime() + 86400000);
|
||||
SelfSignedCertificate ssc = new SelfSignedCertificate("localhost", begin, end);
|
||||
ssl = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey());
|
||||
} else {
|
||||
ssl = SslHelper.getSslContextBuilder(sslServerProperties);
|
||||
}
|
||||
// Can't set to REQUIRE because the CORS pre-flight requests will fail.
|
||||
ssl.clientAuth(ClientAuth.OPTIONAL);
|
||||
return ssl.build();
|
||||
}
|
||||
|
||||
protected SSLContext createSSLClientContext(SslClientProperties sslClientProperties) {
|
||||
|
||||
SslClientProperties ssl = sslClientProperties;
|
||||
return HttpClient.getSSLContext(ssl.getTrustStoreFile(), ssl.getTrustStoreType(), ssl.getTrustStorePassword(), ssl.getKeyStoreFile(),
|
||||
ssl.getKeyStoreType(), ssl.getKeyStorePassword());
|
||||
}
|
||||
|
||||
private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
|
||||
for (String s : paths) {
|
||||
try {
|
||||
Stat stat = curatorFramework.checkExists().forPath(s);
|
||||
if (stat == null) {
|
||||
curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(s);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.info(e.getMessage());
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
protected ChannelHandler setupHttpChannel(BalancerHttpProperties balancerHttpProperties, SslServerProperties sslServerProperties, SslContext sslCtx,
|
||||
MetricResolver metricResolver, HttpClientPool httpClientPool) {
|
||||
|
||||
return new ChannelInitializer<SocketChannel>() {
|
||||
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) {
|
||||
|
||||
ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerHttpProperties, sslCtx));
|
||||
ch.pipeline().addLast("encoder", new HttpResponseEncoder());
|
||||
ch.pipeline().addLast("decoder", new HttpRequestDecoder(4096, 32768, 8192, true, 128));
|
||||
ch.pipeline().addLast("compressor", new HttpContentCompressor());
|
||||
ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
|
||||
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
|
||||
ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
|
||||
ch.pipeline().addLast("queryDecoder",
|
||||
new timely.netty.http.HttpRequestDecoder(authenticationService, securityProperties, balancerHttpProperties));
|
||||
ch.pipeline().addLast("fileServer",
|
||||
new HttpStaticFileServerHandler().setIgnoreSslHandshakeErrors(sslServerProperties.isIgnoreSslHandshakeErrors()));
|
||||
ch.pipeline().addLast("login", new X509LoginRequestHandler(authenticationService, securityProperties, balancerHttpProperties));
|
||||
ch.pipeline().addLast("httpRelay", new HttpRelayHandler(metricResolver, httpClientPool));
|
||||
ch.pipeline().addLast("error", new TimelyExceptionHandler().setIgnoreSslHandshakeErrors(sslServerProperties.isIgnoreSslHandshakeErrors()));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
protected ChannelHandler setupUdpChannel(MetricResolver metricResolver, UdpClientPool udpClientPool) {
|
||||
return new ChannelInitializer<DatagramChannel>() {
|
||||
|
||||
@Override
|
||||
protected void initChannel(DatagramChannel ch) throws Exception {
|
||||
ch.pipeline().addLast("logger", new LoggingHandler());
|
||||
ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
|
||||
ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
|
||||
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
|
||||
ch.pipeline().addLast("putDecoder", new UdpDecoder());
|
||||
ch.pipeline().addLast("udpRelayHandler", new UdpRelayHandler(metricResolver, udpClientPool));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
protected ChannelHandler setupTcpChannel(MetricResolver metricResolver, List<TcpClientPool> tcpClientPools) {
|
||||
return new ChannelInitializer<SocketChannel>() {
|
||||
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) throws Exception {
|
||||
ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
|
||||
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
|
||||
ch.pipeline().addLast("putDecoder", new TcpDecoder());
|
||||
ch.pipeline().addLast("tcpRelayHandler", new TcpRelayHandler(metricResolver, tcpClientPools));
|
||||
ch.pipeline().addLast("versionHandler", new TcpVersionHandler());
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
protected ChannelHandler setupWSChannel(BalancerWebsocketProperties balancerWebsocketProperties, SecurityProperties securityProperties, SslContext sslCtx,
|
||||
MetricResolver metricResolver, WsClientPool wsClientPool) {
|
||||
return new ChannelInitializer<SocketChannel>() {
|
||||
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) throws Exception {
|
||||
ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
|
||||
ch.pipeline().addLast("httpServer", new HttpServerCodec());
|
||||
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
|
||||
ch.pipeline().addLast("sessionExtractor", new WebSocketFullRequestHandler(authenticationService));
|
||||
|
||||
ch.pipeline().addLast("idle-handler", new IdleStateHandler(balancerWebsocketProperties.getTimeout(), 0, 0));
|
||||
ch.pipeline().addLast("ws-protocol", new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, false, true));
|
||||
ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(authenticationService, securityProperties));
|
||||
ch.pipeline().addLast("httpRelay", new WsRelayHandler(balancerWebsocketProperties, metricResolver, wsClientPool));
|
||||
ch.pipeline().addLast("error", new WSTimelyExceptionHandler());
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -1,81 +0,0 @@
|
||||
package timely.balancer.component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.PreDestroy;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.configuration.BalancerServerProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.tcp.TcpClientPool;
|
||||
import timely.balancer.healthcheck.TCPServerHealthCheck;
|
||||
|
||||
@Component
|
||||
@ConditionalOnMissingBean(HealthChecker.class)
|
||||
public class HealthChecker {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(HealthChecker.class);
|
||||
private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
|
||||
private TCPServerHealthCheck tcpServerHealthCheck;
|
||||
private BalancerProperties balancerProperties;
|
||||
private BalancerServerProperties balancerServerProperties;
|
||||
protected List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
|
||||
|
||||
public HealthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
|
||||
this.balancerProperties = balancerProperties;
|
||||
this.balancerServerProperties = balancerServerProperties;
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
public void setup() {
|
||||
TcpClientPool tcpClientPool = new TcpClientPool(balancerServerProperties);
|
||||
this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
|
||||
this.executorService.scheduleAtFixedRate(() -> {
|
||||
synchronized (timelyHosts) {
|
||||
for (TimelyBalancedHost h : timelyHosts) {
|
||||
try {
|
||||
if (tcpServerHealthCheck.isServerHealthy(h)) {
|
||||
h.reportSuccess();
|
||||
} else {
|
||||
h.reportFailure();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage(), e);
|
||||
h.reportFailure();
|
||||
}
|
||||
}
|
||||
}
|
||||
}, 0, balancerProperties.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
public void close() throws Exception {
|
||||
this.executorService.shutdown();
|
||||
try {
|
||||
this.executorService.awaitTermination(5, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException e) {
|
||||
|
||||
} finally {
|
||||
if (!this.executorService.isTerminated()) {
|
||||
this.executorService.shutdownNow();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
|
||||
synchronized (timelyHosts) {
|
||||
this.timelyHosts.clear();
|
||||
this.timelyHosts.addAll(timelyHosts);
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -1,10 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({BalancerProperties.class, BalancerServerProperties.class, BalancerWebsocketProperties.class, BalancerHttpProperties.class})
|
||||
public class BalancerConfiguration {
|
||||
|
||||
}
|
||||
@@ -1,24 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import javax.validation.Valid;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.boot.context.properties.NestedConfigurationProperty;
|
||||
import org.springframework.cloud.context.config.annotation.RefreshScope;
|
||||
|
||||
import timely.common.configuration.HttpProperties;
|
||||
|
||||
@RefreshScope
|
||||
@ConfigurationProperties(prefix = "timely.balancer.http")
|
||||
public class BalancerHttpProperties extends HttpProperties {
|
||||
|
||||
@Valid
|
||||
@NestedConfigurationProperty
|
||||
private GenericKeyedObjectPoolConfig<CloseableHttpClient> httpClientPool = new GenericKeyedObjectPoolConfig<>();
|
||||
|
||||
public GenericKeyedObjectPoolConfig<CloseableHttpClient> getHttpClientPool() {
|
||||
return httpClientPool;
|
||||
}
|
||||
}
|
||||
@@ -1,71 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
@Validated
|
||||
@Component
|
||||
@ConfigurationProperties(prefix = "timely.balancer")
|
||||
public class BalancerProperties {
|
||||
|
||||
private boolean loginRequired = false;
|
||||
|
||||
private String assignmentFile;
|
||||
|
||||
private Double controlBandPercentage = 0.05;
|
||||
|
||||
private long checkServerHealthInterval = 10000;
|
||||
|
||||
private int serverFailuresBeforeDown = 3;
|
||||
|
||||
private int serverSuccessesBeforeUp = 3;
|
||||
|
||||
public void setLoginRequired(boolean loginRequired) {
|
||||
this.loginRequired = loginRequired;
|
||||
}
|
||||
|
||||
public boolean isLoginRequired() {
|
||||
return loginRequired;
|
||||
}
|
||||
|
||||
public String getAssignmentFile() {
|
||||
return assignmentFile;
|
||||
}
|
||||
|
||||
public void setAssignmentFile(String assignmentFile) {
|
||||
this.assignmentFile = assignmentFile;
|
||||
}
|
||||
|
||||
public void setControlBandPercentage(Double controlBandPercentage) {
|
||||
this.controlBandPercentage = controlBandPercentage;
|
||||
}
|
||||
|
||||
public Double getControlBandPercentage() {
|
||||
return controlBandPercentage;
|
||||
}
|
||||
|
||||
public void setCheckServerHealthInterval(long checkServerHealthInterval) {
|
||||
this.checkServerHealthInterval = checkServerHealthInterval;
|
||||
}
|
||||
|
||||
public long getCheckServerHealthInterval() {
|
||||
return checkServerHealthInterval;
|
||||
}
|
||||
|
||||
public int getServerFailuresBeforeDown() {
|
||||
return serverFailuresBeforeDown;
|
||||
}
|
||||
|
||||
public void setServerFailuresBeforeDown(int serverFailuresBeforeDown) {
|
||||
this.serverFailuresBeforeDown = serverFailuresBeforeDown;
|
||||
}
|
||||
|
||||
public int getServerSuccessesBeforeUp() {
|
||||
return serverSuccessesBeforeUp;
|
||||
}
|
||||
|
||||
public void setServerSuccessesBeforeUp(int serverSuccessesBeforeUp) {
|
||||
this.serverSuccessesBeforeUp = serverSuccessesBeforeUp;
|
||||
}
|
||||
}
|
||||
@@ -1,53 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import javax.validation.Valid;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.boot.context.properties.NestedConfigurationProperty;
|
||||
import org.springframework.cloud.context.config.annotation.RefreshScope;
|
||||
|
||||
import timely.client.tcp.TcpClient;
|
||||
import timely.client.udp.UdpClient;
|
||||
import timely.common.configuration.ServerProperties;
|
||||
|
||||
@RefreshScope
|
||||
@ConfigurationProperties(prefix = "timely.balancer.server")
|
||||
public class BalancerServerProperties extends ServerProperties {
|
||||
|
||||
@Valid
|
||||
@NestedConfigurationProperty
|
||||
private GenericKeyedObjectPoolConfig<UdpClient> udpClientPool = new GenericKeyedObjectPoolConfig<>();
|
||||
|
||||
@Valid
|
||||
@NestedConfigurationProperty
|
||||
private GenericKeyedObjectPoolConfig<TcpClient> tcpClientPool = new GenericKeyedObjectPoolConfig<>();
|
||||
|
||||
private int numTcpPools = 1;
|
||||
|
||||
private int tcpBufferSize = -1;
|
||||
|
||||
public GenericKeyedObjectPoolConfig<UdpClient> getUdpClientPool() {
|
||||
return udpClientPool;
|
||||
}
|
||||
|
||||
public GenericKeyedObjectPoolConfig<TcpClient> getTcpClientPool() {
|
||||
return tcpClientPool;
|
||||
}
|
||||
|
||||
public int getNumTcpPools() {
|
||||
return numTcpPools;
|
||||
}
|
||||
|
||||
public void setNumTcpPools(int numTcpPools) {
|
||||
this.numTcpPools = numTcpPools;
|
||||
}
|
||||
|
||||
public int getTcpBufferSize() {
|
||||
return tcpBufferSize;
|
||||
}
|
||||
|
||||
public void setTcpBufferSize(int tcpBufferSize) {
|
||||
this.tcpBufferSize = tcpBufferSize;
|
||||
}
|
||||
}
|
||||
@@ -1,34 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import javax.validation.Valid;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.boot.context.properties.NestedConfigurationProperty;
|
||||
import org.springframework.cloud.context.config.annotation.RefreshScope;
|
||||
|
||||
import timely.client.websocket.subscription.WebSocketSubscriptionClient;
|
||||
import timely.common.configuration.WebsocketProperties;
|
||||
|
||||
@RefreshScope
|
||||
@ConfigurationProperties(prefix = "timely.balancer.websocket")
|
||||
public class BalancerWebsocketProperties extends WebsocketProperties {
|
||||
|
||||
@Valid
|
||||
@NestedConfigurationProperty
|
||||
private GenericKeyedObjectPoolConfig<WebSocketSubscriptionClient> wsClientPool = new GenericKeyedObjectPoolConfig<>();
|
||||
|
||||
public GenericKeyedObjectPoolConfig<WebSocketSubscriptionClient> getWsClientPool() {
|
||||
return wsClientPool;
|
||||
}
|
||||
|
||||
private int incomingBufferSize = 50000000;
|
||||
|
||||
public int getIncomingBufferSize() {
|
||||
return incomingBufferSize;
|
||||
}
|
||||
|
||||
public void setIncomingBufferSize(int incomingBufferSize) {
|
||||
this.incomingBufferSize = incomingBufferSize;
|
||||
}
|
||||
}
|
||||
@@ -1,6 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication(scanBasePackageClasses = {SpringBootstrap.class})
|
||||
public class SpringBootstrap {}
|
||||
@@ -1,182 +0,0 @@
|
||||
package timely.balancer.connection;
|
||||
|
||||
import org.apache.commons.lang3.builder.EqualsBuilder;
|
||||
import org.apache.commons.lang3.builder.HashCodeBuilder;
|
||||
import org.apache.commons.lang3.builder.ToStringBuilder;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import timely.balancer.ArrivalRate;
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
|
||||
public class TimelyBalancedHost {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(TimelyBalancedHost.class);
|
||||
|
||||
private String host;
|
||||
private int tcpPort;
|
||||
private int httpPort;
|
||||
private int wsPort;
|
||||
private int udpPort;
|
||||
private boolean isUp = true;
|
||||
private BalancerProperties balancerProperties;
|
||||
private int failures = 0;
|
||||
private int successes = 0;
|
||||
private ArrivalRate arrivalRate = null;
|
||||
|
||||
private TimelyBalancedHost() {
|
||||
|
||||
}
|
||||
|
||||
private TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort, ArrivalRate arrivalRate) {
|
||||
this.host = host;
|
||||
this.tcpPort = tcpPort;
|
||||
this.httpPort = httpPort;
|
||||
this.wsPort = wsPort;
|
||||
this.udpPort = udpPort;
|
||||
this.arrivalRate = arrivalRate;
|
||||
}
|
||||
|
||||
static public TimelyBalancedHost of(String host, int tcpPort, int httpPort, int wsPort, int udpPort, ArrivalRate arrivalRate) {
|
||||
return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort, arrivalRate);
|
||||
}
|
||||
|
||||
public void setBalancerProperties(BalancerProperties balancerProperties) {
|
||||
this.balancerProperties = balancerProperties;
|
||||
}
|
||||
|
||||
synchronized public boolean isUp() {
|
||||
return isUp;
|
||||
}
|
||||
|
||||
public void reportSuccess() {
|
||||
int serverSuccessesBeforeUp = balancerProperties.getServerSuccessesBeforeUp();
|
||||
String h = host;
|
||||
int p = tcpPort;
|
||||
synchronized (this) {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("success reported host:{} port:{} isUp:{} successes:{} serverSuccessesBeforeUp:{}", h, p, isUp, (successes + 1),
|
||||
serverSuccessesBeforeUp);
|
||||
}
|
||||
if (!isUp) {
|
||||
if (++successes >= serverSuccessesBeforeUp) {
|
||||
isUp = true;
|
||||
successes = 0;
|
||||
log.info("host up host:{} port:{}", h, p);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void reportFailure() {
|
||||
int serverFailuresBeforeDown = balancerProperties.getServerFailuresBeforeDown();
|
||||
String h = host;
|
||||
int p = tcpPort;
|
||||
synchronized (this) {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("failure reported host:{} port:{} isUp:{} failures:{} serverFailuresBeforeUp:{}", h, p, isUp, (failures + 1),
|
||||
serverFailuresBeforeDown);
|
||||
}
|
||||
if (isUp) {
|
||||
if (++failures >= serverFailuresBeforeDown) {
|
||||
isUp = false;
|
||||
failures = 0;
|
||||
log.info("host down host:{} port:{}", h, p);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public String getHost() {
|
||||
return host;
|
||||
}
|
||||
|
||||
public void setHost(String host) {
|
||||
this.host = host;
|
||||
}
|
||||
|
||||
public int getUdpPort() {
|
||||
return udpPort;
|
||||
}
|
||||
|
||||
public void setUdpPort(int udpPort) {
|
||||
this.udpPort = udpPort;
|
||||
}
|
||||
|
||||
public int getTcpPort() {
|
||||
return tcpPort;
|
||||
}
|
||||
|
||||
public void setTcpPort(int tcpPort) {
|
||||
this.tcpPort = tcpPort;
|
||||
}
|
||||
|
||||
public int getHttpPort() {
|
||||
return httpPort;
|
||||
}
|
||||
|
||||
public void setHttpPort(int httpPort) {
|
||||
this.httpPort = httpPort;
|
||||
}
|
||||
|
||||
public int getWsPort() {
|
||||
return wsPort;
|
||||
}
|
||||
|
||||
public void setWsPort(int wsPort) {
|
||||
this.wsPort = wsPort;
|
||||
}
|
||||
|
||||
public void arrived() {
|
||||
if (this.arrivalRate != null) {
|
||||
this.arrivalRate.arrived();
|
||||
}
|
||||
}
|
||||
|
||||
public double getArrivalRate() {
|
||||
if (this.arrivalRate == null) {
|
||||
return 0;
|
||||
} else {
|
||||
return this.arrivalRate.getRate();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
HashCodeBuilder builder = new HashCodeBuilder();
|
||||
builder.append(host);
|
||||
builder.append(tcpPort);
|
||||
builder.append(httpPort);
|
||||
builder.append(wsPort);
|
||||
builder.append(udpPort);
|
||||
return builder.toHashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (o instanceof TimelyBalancedHost) {
|
||||
TimelyBalancedHost other = (TimelyBalancedHost) o;
|
||||
EqualsBuilder builder = new EqualsBuilder();
|
||||
builder.append(host, other.host);
|
||||
builder.append(tcpPort, other.tcpPort);
|
||||
builder.append(httpPort, other.httpPort);
|
||||
builder.append(wsPort, other.wsPort);
|
||||
builder.append(udpPort, other.udpPort);
|
||||
return builder.isEquals();
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
ToStringBuilder builder = new ToStringBuilder(this);
|
||||
builder.append(host);
|
||||
builder.append(tcpPort);
|
||||
builder.append(httpPort);
|
||||
builder.append(wsPort);
|
||||
builder.append(udpPort);
|
||||
builder.append(isUp);
|
||||
return builder.toString();
|
||||
}
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
package timely.balancer.connection.http;
|
||||
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import org.apache.commons.pool2.KeyedPooledObjectFactory;
|
||||
import org.apache.commons.pool2.PooledObject;
|
||||
import org.apache.commons.pool2.impl.DefaultPooledObject;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.common.configuration.SecurityProperties;
|
||||
import timely.common.configuration.SslClientProperties;
|
||||
|
||||
public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,CloseableHttpClient> {
|
||||
|
||||
private final SSLContext sslContext;
|
||||
private final SecurityProperties security;
|
||||
private SslClientProperties sslClientProperties;
|
||||
|
||||
public HttpClientFactory(SecurityProperties security, SslClientProperties sslClientProperties, SSLContext sslContext) {
|
||||
this.security = security;
|
||||
this.sslContext = sslContext;
|
||||
this.sslClientProperties = sslClientProperties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public PooledObject<CloseableHttpClient> makeObject(TimelyBalancedHost k) throws Exception {
|
||||
// disable cookie management because we are sharing connections
|
||||
return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, null, sslClientProperties.isHostVerificationEnabled(),
|
||||
sslClientProperties.isUseClientCert()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroyObject(TimelyBalancedHost k, PooledObject<CloseableHttpClient> o) throws Exception {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean validateObject(TimelyBalancedHost k, PooledObject<CloseableHttpClient> o) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void activateObject(TimelyBalancedHost k, PooledObject<CloseableHttpClient> o) throws Exception {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void passivateObject(TimelyBalancedHost k, PooledObject<CloseableHttpClient> o) throws Exception {
|
||||
|
||||
}
|
||||
}
|
||||
@@ -1,18 +0,0 @@
|
||||
package timely.balancer.connection.http;
|
||||
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
|
||||
import timely.balancer.configuration.BalancerHttpProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.common.configuration.SecurityProperties;
|
||||
import timely.common.configuration.SslClientProperties;
|
||||
|
||||
public class HttpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,CloseableHttpClient> {
|
||||
|
||||
public HttpClientPool(SecurityProperties securityProperties, SslClientProperties sslClientProperties, BalancerHttpProperties http, SSLContext sslContext) {
|
||||
super(new HttpClientFactory(securityProperties, sslClientProperties, sslContext), http.getHttpClientPool());
|
||||
}
|
||||
}
|
||||
@@ -1,57 +0,0 @@
|
||||
package timely.balancer.connection.tcp;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.commons.pool2.KeyedPooledObjectFactory;
|
||||
import org.apache.commons.pool2.PooledObject;
|
||||
import org.apache.commons.pool2.impl.DefaultPooledObject;
|
||||
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.client.tcp.TcpClient;
|
||||
|
||||
public class TcpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,TcpClient> {
|
||||
|
||||
private int bufferSize;
|
||||
|
||||
public TcpClientFactory(int bufferSize) {
|
||||
this.bufferSize = bufferSize;
|
||||
}
|
||||
|
||||
@Override
|
||||
public PooledObject<TcpClient> makeObject(TimelyBalancedHost k) throws Exception {
|
||||
return new DefaultPooledObject<>(new TcpClient(k.getHost(), k.getTcpPort(), bufferSize));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroyObject(TimelyBalancedHost k, PooledObject<TcpClient> o) throws Exception {
|
||||
try {
|
||||
o.getObject().flush();
|
||||
o.getObject().close();
|
||||
} catch (IOException e) {
|
||||
throw new IOException("Error closing connection to " + k.getHost() + ":" + k.getTcpPort());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean validateObject(TimelyBalancedHost k, PooledObject<TcpClient> o) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void activateObject(TimelyBalancedHost k, PooledObject<TcpClient> o) throws Exception {
|
||||
try {
|
||||
o.getObject().open();
|
||||
} catch (Exception e) {
|
||||
throw new IOException("Unable to connect to " + k.getHost() + ":" + k.getTcpPort(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void passivateObject(TimelyBalancedHost k, PooledObject<TcpClient> o) throws Exception {
|
||||
try {
|
||||
o.getObject().flush();
|
||||
} catch (Exception e) {
|
||||
throw new IOException("Error flushing connection to " + k.getHost() + ":" + k.getTcpPort());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,14 +0,0 @@
|
||||
package timely.balancer.connection.tcp;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
|
||||
|
||||
import timely.balancer.configuration.BalancerServerProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.client.tcp.TcpClient;
|
||||
|
||||
public class TcpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,TcpClient> {
|
||||
|
||||
public TcpClientPool(BalancerServerProperties balancerServerProperties) {
|
||||
super(new TcpClientFactory(balancerServerProperties.getTcpBufferSize()), balancerServerProperties.getTcpClientPool());
|
||||
}
|
||||
}
|
||||
@@ -1,53 +0,0 @@
|
||||
package timely.balancer.connection.udp;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.commons.pool2.KeyedPooledObjectFactory;
|
||||
import org.apache.commons.pool2.PooledObject;
|
||||
import org.apache.commons.pool2.impl.DefaultPooledObject;
|
||||
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.client.udp.UdpClient;
|
||||
|
||||
public class UdpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,UdpClient> {
|
||||
|
||||
public UdpClientFactory() {}
|
||||
|
||||
@Override
|
||||
public PooledObject<UdpClient> makeObject(TimelyBalancedHost k) throws Exception {
|
||||
return new DefaultPooledObject<>(new UdpClient(k.getHost(), k.getUdpPort()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroyObject(TimelyBalancedHost k, PooledObject<UdpClient> o) throws Exception {
|
||||
try {
|
||||
o.getObject().flush();
|
||||
o.getObject().close();
|
||||
} catch (IOException e) {
|
||||
throw new IOException("Error closing connection to " + k.getHost() + ":" + k.getUdpPort());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean validateObject(TimelyBalancedHost k, PooledObject<UdpClient> o) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void activateObject(TimelyBalancedHost k, PooledObject<UdpClient> o) throws Exception {
|
||||
try {
|
||||
o.getObject().open();
|
||||
} catch (Exception e) {
|
||||
throw new IOException("Unable to connect to " + k.getHost() + ":" + k.getUdpPort(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void passivateObject(TimelyBalancedHost k, PooledObject<UdpClient> o) throws Exception {
|
||||
try {
|
||||
o.getObject().flush();
|
||||
} catch (Exception e) {
|
||||
throw new IOException("Error flushing connection to " + k.getHost() + ":" + k.getUdpPort());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,14 +0,0 @@
|
||||
package timely.balancer.connection.udp;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
|
||||
|
||||
import timely.balancer.configuration.BalancerServerProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.client.udp.UdpClient;
|
||||
|
||||
public class UdpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,UdpClient> {
|
||||
|
||||
public UdpClientPool(BalancerServerProperties serverProperties) {
|
||||
super(new UdpClientFactory(), serverProperties.getUdpClientPool());
|
||||
}
|
||||
}
|
||||
@@ -1,59 +0,0 @@
|
||||
package timely.balancer.connection.ws;
|
||||
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import org.apache.commons.pool2.KeyedPooledObjectFactory;
|
||||
import org.apache.commons.pool2.PooledObject;
|
||||
import org.apache.commons.pool2.impl.DefaultPooledObject;
|
||||
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.configuration.BalancerWebsocketProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.client.websocket.subscription.WebSocketSubscriptionClient;
|
||||
import timely.common.configuration.SslClientProperties;
|
||||
|
||||
public class WsClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,WebSocketSubscriptionClient> {
|
||||
|
||||
private final SSLContext sslContext;
|
||||
private BalancerProperties balancerProperties;
|
||||
private BalancerWebsocketProperties balancerWebsocketProperties;
|
||||
private SslClientProperties sslClientProperties;
|
||||
|
||||
public WsClientFactory(BalancerProperties balancerProperties, BalancerWebsocketProperties balancerWebsocketProperties,
|
||||
SslClientProperties sslClientProperties, SSLContext sslContext) {
|
||||
this.balancerProperties = balancerProperties;
|
||||
this.balancerWebsocketProperties = balancerWebsocketProperties;
|
||||
this.sslClientProperties = sslClientProperties;
|
||||
this.sslContext = sslContext;
|
||||
}
|
||||
|
||||
@Override
|
||||
public PooledObject<WebSocketSubscriptionClient> makeObject(TimelyBalancedHost k) throws Exception {
|
||||
|
||||
int bufferSize = balancerWebsocketProperties.getIncomingBufferSize();
|
||||
WebSocketSubscriptionClient client = new WebSocketSubscriptionClient(sslContext, k.getHost(), k.getHttpPort(), k.getWsPort(),
|
||||
sslClientProperties.isUseClientCert(), balancerProperties.isLoginRequired(), sslClientProperties.isHostVerificationEnabled(),
|
||||
bufferSize);
|
||||
return new DefaultPooledObject<>(client);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroyObject(TimelyBalancedHost k, PooledObject<WebSocketSubscriptionClient> o) throws Exception {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean validateObject(TimelyBalancedHost k, PooledObject<WebSocketSubscriptionClient> o) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void activateObject(TimelyBalancedHost k, PooledObject<WebSocketSubscriptionClient> o) throws Exception {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void passivateObject(TimelyBalancedHost k, PooledObject<WebSocketSubscriptionClient> o) throws Exception {
|
||||
|
||||
}
|
||||
}
|
||||
@@ -1,20 +0,0 @@
|
||||
package timely.balancer.connection.ws;
|
||||
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
|
||||
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.configuration.BalancerWebsocketProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.client.websocket.subscription.WebSocketSubscriptionClient;
|
||||
import timely.common.configuration.SslClientProperties;
|
||||
|
||||
public class WsClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,WebSocketSubscriptionClient> {
|
||||
|
||||
public WsClientPool(BalancerProperties balancerProperties, BalancerWebsocketProperties balancerWebsocketProperties, SslClientProperties sslClientProperties,
|
||||
SSLContext sslContext) {
|
||||
super(new WsClientFactory(balancerProperties, balancerWebsocketProperties, sslClientProperties, sslContext),
|
||||
balancerWebsocketProperties.getWsClientPool());
|
||||
}
|
||||
}
|
||||
@@ -1,8 +0,0 @@
|
||||
package timely.balancer.healthcheck;
|
||||
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
|
||||
public interface ServerHealthCheck {
|
||||
|
||||
public boolean isServerHealthy(TimelyBalancedHost timelyBalancedHost);
|
||||
}
|
||||
@@ -1,33 +0,0 @@
|
||||
package timely.balancer.healthcheck;
|
||||
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.tcp.TcpClientPool;
|
||||
import timely.client.tcp.TcpClient;
|
||||
|
||||
public class TCPServerHealthCheck implements ServerHealthCheck {
|
||||
|
||||
final private TcpClientPool tcpClientPool;
|
||||
|
||||
public TCPServerHealthCheck(TcpClientPool tcpClientPool) {
|
||||
this.tcpClientPool = tcpClientPool;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isServerHealthy(TimelyBalancedHost timelyBalancedHost) {
|
||||
|
||||
boolean healthy = true;
|
||||
TcpClient client = null;
|
||||
try {
|
||||
client = tcpClientPool.borrowObject(timelyBalancedHost);
|
||||
client.open();
|
||||
} catch (Exception e) {
|
||||
healthy = false;
|
||||
|
||||
} finally {
|
||||
if (client != null) {
|
||||
tcpClientPool.returnObject(timelyBalancedHost, client);
|
||||
}
|
||||
}
|
||||
return healthy;
|
||||
}
|
||||
}
|
||||
@@ -1,203 +0,0 @@
|
||||
package timely.balancer.netty.http;
|
||||
|
||||
import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.net.URLEncoder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.http.Header;
|
||||
import org.apache.http.client.methods.CloseableHttpResponse;
|
||||
import org.apache.http.client.methods.HttpGet;
|
||||
import org.apache.http.client.methods.HttpPost;
|
||||
import org.apache.http.client.methods.HttpUriRequest;
|
||||
import org.apache.http.entity.StringEntity;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.apache.http.message.BasicHeader;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.util.StreamUtils;
|
||||
|
||||
import com.google.common.collect.Multimap;
|
||||
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import io.netty.handler.codec.http.DefaultFullHttpResponse;
|
||||
import io.netty.handler.codec.http.FullHttpRequest;
|
||||
import io.netty.handler.codec.http.FullHttpResponse;
|
||||
import io.netty.handler.codec.http.HttpMethod;
|
||||
import io.netty.handler.codec.http.HttpResponseStatus;
|
||||
import io.netty.handler.codec.http.HttpVersion;
|
||||
import io.netty.util.ReferenceCountUtil;
|
||||
import timely.api.request.AuthenticatedRequest;
|
||||
import timely.api.request.HttpRequest;
|
||||
import timely.api.request.MetricRequest;
|
||||
import timely.api.request.timeseries.QueryRequest;
|
||||
import timely.api.response.TimelyException;
|
||||
import timely.auth.TimelyAuthenticationToken;
|
||||
import timely.auth.util.HttpHeaderUtils;
|
||||
import timely.auth.util.ProxiedEntityUtils;
|
||||
import timely.balancer.component.MetricResolver;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.http.HttpClientPool;
|
||||
import timely.netty.http.TimelyHttpHandler;
|
||||
|
||||
public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> implements TimelyHttpHandler {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(HttpRelayHandler.class);
|
||||
private final HttpClientPool httpClientPool;
|
||||
private MetricResolver metricResolver;
|
||||
|
||||
public HttpRelayHandler(MetricResolver metricResolver, HttpClientPool httpClientPool) {
|
||||
this.metricResolver = metricResolver;
|
||||
this.httpClientPool = httpClientPool;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
|
||||
|
||||
CloseableHttpClient client = null;
|
||||
TimelyBalancedHost k = null;
|
||||
CloseableHttpResponse relayedResponse = null;
|
||||
FullHttpRequest request = null;
|
||||
try {
|
||||
try {
|
||||
request = msg.getHttpRequest();
|
||||
String originalURI = request.uri();
|
||||
originalURI = encodeURI(originalURI);
|
||||
|
||||
String metric = null;
|
||||
if (msg instanceof QueryRequest) {
|
||||
Collection<QueryRequest.SubQuery> subqueries = ((QueryRequest) msg).getQueries();
|
||||
if (subqueries != null) {
|
||||
Iterator<QueryRequest.SubQuery> itr = subqueries.iterator();
|
||||
if (itr.hasNext()) {
|
||||
metric = itr.next().getMetric();
|
||||
}
|
||||
}
|
||||
k = metricResolver.getHostPortKey(metric);
|
||||
} else if (msg instanceof MetricRequest) {
|
||||
metric = ((MetricRequest) msg).getMetric().getName();
|
||||
k = metricResolver.getHostPortKeyIngest(metric);
|
||||
} else {
|
||||
k = metricResolver.getHostPortKey(null);
|
||||
}
|
||||
client = httpClientPool.borrowObject(k);
|
||||
|
||||
Multimap<String,String> headers = HttpHeaderUtils.toMultimap(request.headers());
|
||||
if (msg instanceof AuthenticatedRequest) {
|
||||
AuthenticatedRequest authenticatedRequest = (AuthenticatedRequest) msg;
|
||||
TimelyAuthenticationToken token = authenticatedRequest.getToken();
|
||||
if (token != null) {
|
||||
if (token.getClientCert() != null) {
|
||||
ProxiedEntityUtils.addProxyHeaders(headers, token.getClientCert());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
List<Header> relayedHeaderList = new ArrayList<>();
|
||||
for (Map.Entry<String,String> h : headers.entries()) {
|
||||
if (!h.getKey().equals(CONTENT_LENGTH.toString())) {
|
||||
relayedHeaderList.add(new BasicHeader(h.getKey(), h.getValue()));
|
||||
}
|
||||
}
|
||||
Header[] headerArray = new Header[relayedHeaderList.size()];
|
||||
relayedHeaderList.toArray(headerArray);
|
||||
|
||||
String relayURI = "https://" + k.getHost() + ":" + k.getHttpPort() + originalURI;
|
||||
if (request.method().equals(HttpMethod.GET)) {
|
||||
HttpUriRequest relayedRequest = new HttpGet(relayURI);
|
||||
if (headerArray.length > 0) {
|
||||
relayedRequest.setHeaders(headerArray);
|
||||
}
|
||||
relayedResponse = client.execute(relayedRequest);
|
||||
|
||||
} else if (request.method().equals(HttpMethod.POST)) {
|
||||
HttpPost relayedRequest = new HttpPost(relayURI);
|
||||
|
||||
String content = request.content().toString(StandardCharsets.UTF_8);
|
||||
StringEntity entity = new StringEntity(content);
|
||||
relayedRequest.setEntity(entity);
|
||||
if (headerArray.length > 0) {
|
||||
relayedRequest.setHeaders(headerArray);
|
||||
}
|
||||
relayedResponse = client.execute(relayedRequest);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
String message = e.getMessage();
|
||||
if (message == null) {
|
||||
log.error("", e);
|
||||
} else {
|
||||
if (message.contains("No matching tags")) {
|
||||
log.trace(message);
|
||||
} else {
|
||||
log.error(message, e);
|
||||
}
|
||||
}
|
||||
this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), e.getMessage(), e.getLocalizedMessage(), e));
|
||||
return;
|
||||
} finally {
|
||||
if (request != null) {
|
||||
ReferenceCountUtil.release(request);
|
||||
}
|
||||
}
|
||||
FullHttpResponse response = null;
|
||||
if (relayedResponse == null) {
|
||||
this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.METHOD_NOT_ALLOWED.code(), "Method not allowed", "Method not allowed"));
|
||||
} else {
|
||||
ByteArrayOutputStream baos = new MyByteArrayOutputStream();
|
||||
StreamUtils.copy(relayedResponse.getEntity().getContent(), baos);
|
||||
|
||||
response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.valueOf(relayedResponse.getStatusLine().getStatusCode()),
|
||||
Unpooled.copiedBuffer(baos.toByteArray()));
|
||||
for (Header h : relayedResponse.getAllHeaders()) {
|
||||
response.headers().add(h.getName(), h.getValue());
|
||||
}
|
||||
response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
|
||||
sendResponse(ctx, response);
|
||||
}
|
||||
} finally {
|
||||
if (relayedResponse != null) {
|
||||
relayedResponse.close();
|
||||
}
|
||||
if (client != null && k != null) {
|
||||
httpClientPool.returnObject(k, client);
|
||||
} else {
|
||||
log.error("NOT RETURNING CONNECTION! " + msg.getHttpRequest().uri());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static public class MyByteArrayOutputStream extends ByteArrayOutputStream {
|
||||
|
||||
public MyByteArrayOutputStream() {}
|
||||
|
||||
public MyByteArrayOutputStream(int size) {
|
||||
super(size);
|
||||
}
|
||||
|
||||
public int getCount() {
|
||||
return count;
|
||||
}
|
||||
|
||||
public byte[] getBuf() {
|
||||
return buf;
|
||||
}
|
||||
}
|
||||
|
||||
static public String encodeURI(String s) {
|
||||
try {
|
||||
s = s.replaceAll("\\{", URLEncoder.encode("{", "UTF-8"));
|
||||
s = s.replaceAll("\\}", URLEncoder.encode("}", "UTF-8"));
|
||||
} catch (Exception e) {
|
||||
|
||||
}
|
||||
return s;
|
||||
}
|
||||
}
|
||||
@@ -1,97 +0,0 @@
|
||||
package timely.balancer.netty.tcp;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.commons.lang3.tuple.Pair;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import timely.api.request.MetricRequest;
|
||||
import timely.api.request.TcpRequest;
|
||||
import timely.balancer.component.MetricResolver;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.tcp.TcpClientPool;
|
||||
import timely.client.tcp.TcpClient;
|
||||
import timely.netty.Constants;
|
||||
|
||||
public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(TcpRelayHandler.class);
|
||||
private static final String LOG_ERR_MSG = "Error storing put metric: {}";
|
||||
private static final String ERR_MSG = "Error storing put metric: ";
|
||||
|
||||
private MetricResolver metricResolver;
|
||||
private List<TcpClientPool> tcpClientPools;
|
||||
|
||||
public TcpRelayHandler(MetricResolver metricResolver, List<TcpClientPool> tcpClientPools) {
|
||||
this.metricResolver = metricResolver;
|
||||
this.tcpClientPools = tcpClientPools;
|
||||
}
|
||||
|
||||
private TcpClientPool tcpClientPool() {
|
||||
int x = Math.toIntExact(Thread.currentThread().getId() % tcpClientPools.size());
|
||||
return tcpClientPools.get(x);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, TcpRequest msg) throws Exception {
|
||||
log.trace("Received {}", msg);
|
||||
try {
|
||||
String line;
|
||||
Pair<TimelyBalancedHost,TcpClient> keyClientPair = null;
|
||||
try {
|
||||
if (msg instanceof MetricRequest) {
|
||||
String metricName = ((MetricRequest) msg).getMetric().getName();
|
||||
line = ((MetricRequest) msg).getLine();
|
||||
keyClientPair = getClient(metricName, true);
|
||||
} else {
|
||||
// Version request
|
||||
line = "version";
|
||||
keyClientPair = getClient(null, false);
|
||||
}
|
||||
keyClientPair.getRight().write(line + "\n");
|
||||
} finally {
|
||||
if (keyClientPair != null && keyClientPair.getLeft() != null && keyClientPair.getRight() != null) {
|
||||
tcpClientPool().returnObject(keyClientPair.getLeft(), keyClientPair.getRight());
|
||||
}
|
||||
}
|
||||
|
||||
} catch (IOException e) {
|
||||
log.error(LOG_ERR_MSG, msg, e);
|
||||
ChannelFuture cf = ctx.writeAndFlush(Unpooled.copiedBuffer((ERR_MSG + e.getMessage() + "\n").getBytes(StandardCharsets.UTF_8)));
|
||||
if (!cf.isSuccess()) {
|
||||
log.error(Constants.ERR_WRITING_RESPONSE, cf.cause());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Pair<TimelyBalancedHost,TcpClient> getClient(String metric, boolean metricRequest) {
|
||||
TcpClient client = null;
|
||||
TimelyBalancedHost k = null;
|
||||
int failures = 0;
|
||||
while (client == null) {
|
||||
try {
|
||||
k = (metricRequest == true) ? metricResolver.getHostPortKeyIngest(metric) : metricResolver.getHostPortKey(metric);
|
||||
client = tcpClientPool().borrowObject(k);
|
||||
} catch (Exception e1) {
|
||||
failures++;
|
||||
client = null;
|
||||
if (failures % 10 == 0) {
|
||||
log.error(e1.getMessage(), e1);
|
||||
}
|
||||
try {
|
||||
Thread.sleep(failures < 10 ? 500 : 60000);
|
||||
} catch (InterruptedException e2) {
|
||||
// nothing
|
||||
}
|
||||
}
|
||||
}
|
||||
return Pair.of(k, client);
|
||||
}
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
package timely.balancer.netty.udp;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import org.apache.commons.lang3.tuple.Pair;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import timely.api.request.MetricRequest;
|
||||
import timely.api.request.UdpRequest;
|
||||
import timely.balancer.component.MetricResolver;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.udp.UdpClientPool;
|
||||
import timely.client.udp.UdpClient;
|
||||
import timely.netty.Constants;
|
||||
|
||||
public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(UdpRelayHandler.class);
|
||||
private static final String LOG_ERR_MSG = "Error storing put metric: {}";
|
||||
private static final String ERR_MSG = "Error storing put metric: ";
|
||||
|
||||
private MetricResolver metricResolver;
|
||||
private UdpClientPool udpClientPool;
|
||||
|
||||
public UdpRelayHandler(MetricResolver metricResolver, UdpClientPool udpClientPool) {
|
||||
this.metricResolver = metricResolver;
|
||||
this.udpClientPool = udpClientPool;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, UdpRequest msg) throws Exception {
|
||||
log.trace("Received {}", msg);
|
||||
try {
|
||||
MetricRequest metricRequest = (MetricRequest) msg;
|
||||
String metricName = metricRequest.getMetric().getName();
|
||||
Pair<TimelyBalancedHost,UdpClient> keyClientPair = null;
|
||||
try {
|
||||
keyClientPair = getClient(metricName, true);
|
||||
keyClientPair.getRight().write(metricRequest.getLine() + "\n");
|
||||
keyClientPair.getRight().flush();
|
||||
} finally {
|
||||
if (keyClientPair != null && keyClientPair.getLeft() != null && keyClientPair.getRight() != null) {
|
||||
udpClientPool.returnObject(keyClientPair.getLeft(), keyClientPair.getRight());
|
||||
}
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error(LOG_ERR_MSG, msg, e);
|
||||
ChannelFuture cf = ctx.writeAndFlush(Unpooled.copiedBuffer((ERR_MSG + e.getMessage() + "\n").getBytes(StandardCharsets.UTF_8)));
|
||||
if (!cf.isSuccess()) {
|
||||
log.error(Constants.ERR_WRITING_RESPONSE, cf.cause());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Pair<TimelyBalancedHost,UdpClient> getClient(String metric, boolean metricRequest) {
|
||||
UdpClient client = null;
|
||||
TimelyBalancedHost k = null;
|
||||
int failures = 0;
|
||||
while (client == null) {
|
||||
try {
|
||||
k = (metricRequest == true) ? metricResolver.getHostPortKeyIngest(metric) : metricResolver.getHostPortKey(metric);
|
||||
client = udpClientPool.borrowObject(k);
|
||||
} catch (Exception e1) {
|
||||
failures++;
|
||||
client = null;
|
||||
log.error(e1.getMessage(), e1);
|
||||
try {
|
||||
Thread.sleep(failures < 10 ? 500 : 60000);
|
||||
} catch (InterruptedException e2) {
|
||||
// nothing
|
||||
}
|
||||
}
|
||||
}
|
||||
return Pair.of(k, client);
|
||||
}
|
||||
}
|
||||
@@ -1,94 +0,0 @@
|
||||
package timely.balancer.netty.ws;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import javax.websocket.CloseReason;
|
||||
import javax.websocket.EndpointConfig;
|
||||
import javax.websocket.Session;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.google.common.collect.HashMultimap;
|
||||
import com.google.common.collect.Multimap;
|
||||
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.http.HttpResponseStatus;
|
||||
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
|
||||
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
|
||||
import io.netty.util.concurrent.ScheduledFuture;
|
||||
import timely.api.response.TimelyException;
|
||||
import timely.auth.TimelyAuthenticationToken;
|
||||
import timely.auth.util.ProxiedEntityUtils;
|
||||
import timely.client.websocket.ClientHandler;
|
||||
|
||||
public class WsClientHandler extends ClientHandler {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(WsClientHandler.class);
|
||||
private final ChannelHandlerContext ctx;
|
||||
private final TimelyAuthenticationToken token;
|
||||
private ScheduledFuture<?> ping;
|
||||
|
||||
public WsClientHandler(ChannelHandlerContext ctx, TimelyAuthenticationToken token, int pingRate) {
|
||||
this.ctx = ctx;
|
||||
this.token = token;
|
||||
this.ping = this.ctx.executor().scheduleAtFixedRate(() -> {
|
||||
log.trace("Sending ping on channel {}", ctx.channel());
|
||||
ctx.writeAndFlush(new PingWebSocketFrame());
|
||||
}, pingRate, pingRate, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void beforeRequest(Map<String,List<String>> headers) {
|
||||
if (token.getClientCert() != null) {
|
||||
Multimap<String,String> proxyRequestHeaders = HashMultimap.create();
|
||||
ProxiedEntityUtils.addProxyHeaders(proxyRequestHeaders, token.getClientCert());
|
||||
for (String s : proxyRequestHeaders.keySet()) {
|
||||
headers.put(s, new ArrayList<>(proxyRequestHeaders.get(s)));
|
||||
}
|
||||
}
|
||||
|
||||
Multimap<String,String> originalRequestHeaders = token.getHttpHeaders();
|
||||
for (String s : originalRequestHeaders.keySet()) {
|
||||
if (!headers.containsKey(s)) {
|
||||
// add pre-existing values if key does not exist in proxyRequest
|
||||
headers.put(s, new ArrayList<>(originalRequestHeaders.get(s)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onOpen(Session session, EndpointConfig config) {
|
||||
session.addMessageHandler(String.class, message -> {
|
||||
ctx.writeAndFlush(new TextWebSocketFrame(message));
|
||||
log.debug("Message received on Websocket session {}: {}", session.getId(), message);
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onClose(Session session, CloseReason reason) {
|
||||
super.onClose(session, reason);
|
||||
if (!reason.getCloseCode().equals(CloseReason.CloseCodes.NORMAL_CLOSURE)) {
|
||||
log.error("Abnormal close: " + reason.getReasonPhrase());
|
||||
Exception e = new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), reason.getReasonPhrase(), "");
|
||||
WsRelayHandler.sendErrorResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e);
|
||||
}
|
||||
if (!ping.isCancelled()) {
|
||||
ping.cancel(false);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Session session, Throwable t) {
|
||||
super.onError(session, t);
|
||||
log.error(t.getMessage(), t);
|
||||
Exception e = new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), t.getMessage(), "");
|
||||
WsRelayHandler.sendErrorResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e);
|
||||
if (!ping.isCancelled()) {
|
||||
ping.cancel(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
package timely.balancer.netty.ws;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.ws.WsClientPool;
|
||||
import timely.client.websocket.subscription.WebSocketSubscriptionClient;
|
||||
|
||||
public class WsClientHolder {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(WsClientHolder.class);
|
||||
private WebSocketSubscriptionClient client;
|
||||
private TimelyBalancedHost host;
|
||||
|
||||
public WsClientHolder(TimelyBalancedHost host, WebSocketSubscriptionClient client) {
|
||||
this.host = host;
|
||||
this.client = client;
|
||||
}
|
||||
|
||||
synchronized public WebSocketSubscriptionClient getClient() {
|
||||
return client;
|
||||
}
|
||||
|
||||
synchronized public TimelyBalancedHost getHost() {
|
||||
return host;
|
||||
}
|
||||
|
||||
synchronized public void close(WsClientPool wsClientPool) {
|
||||
if (client != null && host != null) {
|
||||
try {
|
||||
client.close();
|
||||
} catch (IOException e) {
|
||||
log.error("Error closing web socket client", e);
|
||||
}
|
||||
wsClientPool.returnObject(host, client);
|
||||
client = null;
|
||||
host = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,204 +0,0 @@
|
||||
package timely.balancer.netty.ws;
|
||||
|
||||
import java.net.ConnectException;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.SerializationFeature;
|
||||
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
|
||||
import com.google.common.collect.Multimap;
|
||||
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelFutureListener;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import io.netty.handler.codec.http.HttpResponseStatus;
|
||||
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
|
||||
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
|
||||
import timely.api.request.websocket.AddSubscription;
|
||||
import timely.api.request.websocket.CloseSubscription;
|
||||
import timely.api.request.websocket.CreateSubscription;
|
||||
import timely.api.request.websocket.RemoveSubscription;
|
||||
import timely.api.request.websocket.SubscriptionRequest;
|
||||
import timely.api.response.TimelyException;
|
||||
import timely.auth.util.ProxiedEntityUtils;
|
||||
import timely.balancer.component.MetricResolver;
|
||||
import timely.balancer.configuration.BalancerWebsocketProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.balancer.connection.ws.WsClientPool;
|
||||
import timely.client.websocket.subscription.WebSocketSubscriptionClient;
|
||||
import timely.netty.http.TimelyHttpHandler;
|
||||
|
||||
public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequest> implements TimelyHttpHandler {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(WsRelayHandler.class);
|
||||
|
||||
private BalancerWebsocketProperties balancerWebsocketProperties;
|
||||
private WsClientPool wsClientPool;
|
||||
private MetricResolver metricResolver;
|
||||
static private ObjectMapper mapper;
|
||||
Map<String,Map<String,WsClientHolder>> wsClients = new ConcurrentHashMap<>();
|
||||
Map<String,ChannelHandlerContext> wsSubscriptions = new ConcurrentHashMap<>();
|
||||
|
||||
static {
|
||||
mapper = new ObjectMapper();
|
||||
mapper.registerModule(new Jdk8Module());
|
||||
mapper.configure(SerializationFeature.INDENT_OUTPUT, true);
|
||||
mapper.configure(SerializationFeature.WRAP_EXCEPTIONS, true);
|
||||
mapper.configure(SerializationFeature.WRAP_ROOT_VALUE, true);
|
||||
}
|
||||
|
||||
public WsRelayHandler(BalancerWebsocketProperties balancerWebsocketProperties, MetricResolver metricResolver, WsClientPool wsClientPool) {
|
||||
this.balancerWebsocketProperties = balancerWebsocketProperties;
|
||||
this.metricResolver = metricResolver;
|
||||
this.wsClientPool = wsClientPool;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, SubscriptionRequest msg) throws Exception {
|
||||
TimelyBalancedHost k;
|
||||
ChannelHandlerContext origContext = null;
|
||||
String subscriptionId;
|
||||
try {
|
||||
String metric = null;
|
||||
|
||||
Multimap<String,String> headers = msg.getRequestHeaders();
|
||||
ProxiedEntityUtils.addProxyHeaders(headers, msg.getToken().getClientCert());
|
||||
|
||||
if (msg instanceof CreateSubscription) {
|
||||
CreateSubscription create = (CreateSubscription) msg;
|
||||
final String currentSubscriptionId = create.getSubscriptionId();
|
||||
subscriptionId = currentSubscriptionId;
|
||||
wsSubscriptions.put(subscriptionId, ctx);
|
||||
// WebSocketSubscriptionClient sends the createSubscription when
|
||||
// we call open
|
||||
// We don't know where to send it to yet anyway because that
|
||||
// depends on the metric.
|
||||
ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
|
||||
|
||||
@Override
|
||||
public void operationComplete(ChannelFuture future) throws Exception {
|
||||
Map<String,WsClientHolder> metricToClientMap = wsClients.get(currentSubscriptionId);
|
||||
|
||||
if (metricToClientMap != null) {
|
||||
synchronized (metricToClientMap) {
|
||||
// close all clients for this subscriptionId
|
||||
log.debug("Channel closed, closing subscriptions for subscriptionId:{}", currentSubscriptionId);
|
||||
for (Map.Entry<String,WsClientHolder> entry : metricToClientMap.entrySet()) {
|
||||
entry.getValue().close(wsClientPool);
|
||||
}
|
||||
wsClients.remove(currentSubscriptionId);
|
||||
wsSubscriptions.remove(currentSubscriptionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
} else if (msg instanceof AddSubscription) {
|
||||
AddSubscription add = (AddSubscription) msg;
|
||||
metric = add.getMetric();
|
||||
subscriptionId = add.getSubscriptionId();
|
||||
origContext = wsSubscriptions.get(subscriptionId);
|
||||
if (origContext == null) {
|
||||
// send error because user never sent a CreateSubscription
|
||||
// and we have nowhere to send the results
|
||||
log.info("ChannelHandlerContext not found for subscriptionId:{} - createSubscription not called?", subscriptionId);
|
||||
sendErrorResponse(ctx, HttpResponseStatus.BAD_REQUEST, new IllegalArgumentException("Must call create first"));
|
||||
}
|
||||
Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
|
||||
if (metricToClientMap == null) {
|
||||
metricToClientMap = new HashMap<>();
|
||||
wsClients.put(subscriptionId, metricToClientMap);
|
||||
}
|
||||
WebSocketSubscriptionClient client = null;
|
||||
synchronized (metricToClientMap) {
|
||||
WsClientHolder hostClientHolder = metricToClientMap.get(metric);
|
||||
if (hostClientHolder == null) {
|
||||
k = metricResolver.getHostPortKey(metric);
|
||||
client = wsClientPool.borrowObject(k);
|
||||
client.open(new WsClientHandler(origContext, add.getToken(), (balancerWebsocketProperties.getTimeout() / 2)));
|
||||
metricToClientMap.put(metric, new WsClientHolder(k, client));
|
||||
} else {
|
||||
client = hostClientHolder.getClient();
|
||||
}
|
||||
}
|
||||
Map<String,String> tags = null;
|
||||
|
||||
Long startTime = 0L;
|
||||
Long endTime = 0L;
|
||||
Long delayTime = 5000L;
|
||||
if (add.getTags().isPresent()) {
|
||||
tags = add.getTags().get();
|
||||
}
|
||||
if (add.getStartTime().isPresent()) {
|
||||
startTime = add.getStartTime().get();
|
||||
}
|
||||
if (add.getEndTime().isPresent()) {
|
||||
endTime = add.getEndTime().get();
|
||||
}
|
||||
if (add.getDelayTime().isPresent()) {
|
||||
delayTime = add.getDelayTime().get();
|
||||
}
|
||||
|
||||
client.addSubscription(add.getMetric(), tags, startTime, endTime, delayTime);
|
||||
|
||||
} else if (msg instanceof RemoveSubscription) {
|
||||
RemoveSubscription remove = (RemoveSubscription) msg;
|
||||
metric = remove.getMetric();
|
||||
subscriptionId = remove.getSubscriptionId();
|
||||
Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
|
||||
if (metricToClientMap != null) {
|
||||
synchronized (metricToClientMap) {
|
||||
WsClientHolder hostClientHolder = metricToClientMap.get(metric);
|
||||
if (hostClientHolder != null) {
|
||||
hostClientHolder.getClient().removeSubscription(metric);
|
||||
} else {
|
||||
log.info("client not found for subscriptionId:{} metric:{}", subscriptionId, metric);
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (msg instanceof CloseSubscription) {
|
||||
CloseSubscription close = (CloseSubscription) msg;
|
||||
subscriptionId = close.getSubscriptionId();
|
||||
Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
|
||||
if (metricToClientMap != null) {
|
||||
synchronized (metricToClientMap) {
|
||||
// close all clients for this subscriptionId
|
||||
for (Map.Entry<String,WsClientHolder> entry : metricToClientMap.entrySet()) {
|
||||
entry.getValue().close(wsClientPool);
|
||||
}
|
||||
}
|
||||
wsClients.remove(subscriptionId);
|
||||
wsSubscriptions.remove(subscriptionId);
|
||||
}
|
||||
ctx.writeAndFlush(new CloseWebSocketFrame(1000, "Client requested close."));
|
||||
}
|
||||
} catch (ConnectException e1) {
|
||||
log.error(e1.getMessage(), e1);
|
||||
ChannelHandlerContext errorCtx = origContext == null ? ctx : origContext;
|
||||
sendErrorResponse(errorCtx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e1);
|
||||
} catch (Exception e2) {
|
||||
log.error(e2.getMessage(), e2);
|
||||
ChannelHandlerContext errorCtx = origContext == null ? ctx : origContext;
|
||||
sendErrorResponse(errorCtx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e2);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
static public void sendErrorResponse(ChannelHandlerContext ctx, HttpResponseStatus status, Exception ex) {
|
||||
|
||||
try {
|
||||
TimelyException te = new TimelyException(status.code(), ex.getMessage(), ex.getMessage());
|
||||
String json = mapper.writeValueAsString(te);
|
||||
ctx.writeAndFlush(new TextWebSocketFrame(json));
|
||||
} catch (JsonProcessingException e) {
|
||||
log.error("Error serializing exception");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +0,0 @@
|
||||
spring:
|
||||
application:
|
||||
name: timely-balancer
|
||||
config:
|
||||
name: timely-balancer
|
||||
location: ${CONF_DIR}/
|
||||
@@ -1,21 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<FindBugsFilter>
|
||||
<Match>
|
||||
<Class name="timely.balancer.component.Balancer" />
|
||||
<Method name="shutdown" />
|
||||
<Bug pattern="UC_USELESS_OBJECT" />
|
||||
</Match>
|
||||
<Match>
|
||||
<Class name="timely.balancer.netty.http.HttpRelayHandler$MyByteArrayOutputStream" />
|
||||
<Method name="getBuf" />
|
||||
<Bug pattern="EI_EXPOSE_REP" />
|
||||
</Match>
|
||||
<Match>
|
||||
<!-- these are probably avoidable, but needs further triage to fix -->
|
||||
<Bug pattern="EI_EXPOSE_REP" />
|
||||
</Match>
|
||||
<Match>
|
||||
<!-- these are probably avoidable, but needs further triage to fix -->
|
||||
<Bug pattern="EI_EXPOSE_REP2" />
|
||||
</Match>
|
||||
</FindBugsFilter>
|
||||
@@ -1,33 +0,0 @@
|
||||
package timely.balancer.component;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import timely.balancer.configuration.BalancerHttpProperties;
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.configuration.BalancerServerProperties;
|
||||
import timely.balancer.configuration.BalancerWebsocketProperties;
|
||||
import timely.common.component.AuthenticationService;
|
||||
import timely.common.configuration.SecurityProperties;
|
||||
import timely.common.configuration.SslClientProperties;
|
||||
import timely.common.configuration.SslServerProperties;
|
||||
import timely.common.configuration.ZookeeperProperties;
|
||||
|
||||
@Primary
|
||||
@Component
|
||||
public class TestBalancer extends Balancer {
|
||||
private static final Logger log = LoggerFactory.getLogger(TestBalancer.class);
|
||||
|
||||
public TestBalancer(ApplicationContext applicationContext, AuthenticationService authenticationService, MetricResolver metricResolver,
|
||||
BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
|
||||
SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
|
||||
BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
|
||||
super(applicationContext, authenticationService, metricResolver, balancerProperties, zookeeperProperties, securityProperties, sslServerProperties,
|
||||
sslClientProperties, balancerServerProperties, balancerHttpProperties, balancerWebsocketProperties);
|
||||
DEFAULT_EVENT_LOOP_THREADS = 1;
|
||||
log.info("Starting TestBalancer");
|
||||
}
|
||||
}
|
||||
@@ -1,33 +0,0 @@
|
||||
package timely.balancer.component;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.configuration.BalancerServerProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
|
||||
@Primary
|
||||
@Component
|
||||
public class TestHealthChecker extends HealthChecker {
|
||||
|
||||
public TestHealthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
|
||||
super(balancerProperties, balancerServerProperties);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setup() {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
|
||||
}
|
||||
|
||||
public List<TimelyBalancedHost> getTimelyHosts() {
|
||||
return this.timelyHosts;
|
||||
}
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
package timely.balancer.component;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.fs.FileSystem;
|
||||
import org.apache.hadoop.fs.Path;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import timely.balancer.configuration.BalancerProperties;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.common.configuration.CacheProperties;
|
||||
|
||||
@Primary
|
||||
@Component
|
||||
public class TestMetricResolver extends MetricResolver {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(TestMetricResolver.class);
|
||||
|
||||
public TestMetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
|
||||
HealthChecker healthChecker) {
|
||||
super(curatorFramework, balancerProperties, cacheProperties, healthChecker);
|
||||
}
|
||||
|
||||
protected void createAssignmentFile() {
|
||||
try {
|
||||
log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
|
||||
Configuration configuration = new Configuration();
|
||||
FileSystem fs = FileSystem.get(configuration);
|
||||
Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
|
||||
fs.setWriteChecksum(false);
|
||||
fs.create(assignmentFile);
|
||||
fs.deleteOnExit(assignmentFile);
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
public List<TimelyBalancedHost> getServerList() {
|
||||
return this.serverList;
|
||||
}
|
||||
}
|
||||
@@ -1,20 +0,0 @@
|
||||
package timely.balancer.component;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.security.authentication.ProviderManager;
|
||||
import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Primary
|
||||
@Component
|
||||
public class TestTimelyAuthenticationManager extends ProviderManager {
|
||||
|
||||
@Autowired
|
||||
public TestTimelyAuthenticationManager(@Qualifier("file") PreAuthenticatedAuthenticationProvider file) {
|
||||
super(Collections.singletonList(file));
|
||||
}
|
||||
}
|
||||
@@ -1,21 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import timely.common.configuration.AccumuloProperties;
|
||||
|
||||
@Configuration
|
||||
public class TestConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public AccumuloProperties accumuloProperties() {
|
||||
AccumuloProperties accumuloProperties = new AccumuloProperties();
|
||||
accumuloProperties.setInstanceName("default");
|
||||
accumuloProperties.setUsername("root");
|
||||
accumuloProperties.setPassword("secret");
|
||||
return accumuloProperties;
|
||||
}
|
||||
}
|
||||
@@ -1,49 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
|
||||
|
||||
import org.apache.curator.RetryPolicy;
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.CuratorFrameworkFactory;
|
||||
import org.apache.curator.retry.RetryForever;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.DependsOn;
|
||||
|
||||
import timely.common.configuration.ZookeeperProperties;
|
||||
|
||||
/*
|
||||
* The purpose of this class is to wait for the creation of either ZooKeeperServerMain (in-memory)
|
||||
* or MiniAccumuloCluster (mini accumulo) before creating a CuratorFramework so that the components
|
||||
* that need CuratorFramework autowired will wait.
|
||||
*/
|
||||
@Configuration
|
||||
public class TestCuratorConfiguration {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(TestCuratorConfiguration.class);
|
||||
|
||||
@Bean(destroyMethod = "close")
|
||||
@ConditionalOnProperty(name = "timely.in-memory-accumulo.enabled", havingValue = "true")
|
||||
@DependsOn("ZooKeeperServerMain")
|
||||
public CuratorFramework inMemoryCuratorFramework(ZookeeperProperties zookeeperProperties) {
|
||||
RetryPolicy retryPolicy = new RetryForever(1000);
|
||||
int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
|
||||
CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
|
||||
curatorFramework.start();
|
||||
return curatorFramework;
|
||||
}
|
||||
|
||||
@Bean(destroyMethod = "close")
|
||||
@ConditionalOnProperty(name = "timely.mini-accumulo.enabled", havingValue = "true")
|
||||
@DependsOn("MiniAccumuloCluster")
|
||||
public CuratorFramework miniAccumuloCuratorFramework(ZookeeperProperties zookeeperProperties) {
|
||||
RetryPolicy retryPolicy = new RetryForever(1000);
|
||||
int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
|
||||
CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
|
||||
curatorFramework.start();
|
||||
return curatorFramework;
|
||||
}
|
||||
}
|
||||
@@ -1,23 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import org.apache.accumulo.core.client.AccumuloClient;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
|
||||
import timely.balancer.test.BalancerTestRule;
|
||||
import timely.common.configuration.MiniAccumuloProperties;
|
||||
import timely.common.configuration.TimelyProperties;
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({MiniAccumuloProperties.class, TimelyProperties.class})
|
||||
public class TestWatcherConfiguration {
|
||||
|
||||
@Bean
|
||||
@Scope("prototype")
|
||||
public BalancerTestRule timelyTestWatcherAccumulo(AccumuloClient accumuloClient, TimelyProperties timelyProperties) throws Exception {
|
||||
|
||||
return new BalancerTestRule(timelyProperties, accumuloClient);
|
||||
}
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
package timely.balancer.configuration;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.ServerSocket;
|
||||
import java.nio.file.Files;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
import org.apache.zookeeper.server.ServerConfig;
|
||||
import org.apache.zookeeper.server.ZooKeeperServerMain;
|
||||
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import timely.common.configuration.ZookeeperProperties;
|
||||
|
||||
@Configuration
|
||||
@ConditionalOnProperty(name = "timely.in-memory-accumulo.enabled", havingValue = "true")
|
||||
public class TestZookeeperConfiguration {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(TestZookeeperConfiguration.class);
|
||||
private static final int MAX_PORT = 100;
|
||||
private int port;
|
||||
|
||||
private ZookeeperProperties zookeeperProperties;
|
||||
|
||||
@Bean(name = "ZooKeeperServerMain", destroyMethod = "close")
|
||||
public ZooKeeperServerMain zooKeeperServerMain(ZookeeperProperties zookeeperProperties) throws Exception {
|
||||
this.zookeeperProperties = zookeeperProperties;
|
||||
this.port = getNextPort(2181);
|
||||
this.zookeeperProperties.setServers(String.format("localhost:%d", port));
|
||||
|
||||
File tempDir;
|
||||
try {
|
||||
tempDir = Files.createTempDirectory("zookeeper_temp").toFile();
|
||||
tempDir.deleteOnExit();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException("Unable to create temp directory for mini accumulo cluster");
|
||||
}
|
||||
|
||||
Properties properties = new Properties();
|
||||
properties.put("dataDir", tempDir.getAbsolutePath() + "/data");
|
||||
properties.put("clientPort", port);
|
||||
properties.put("skipACL", "true");
|
||||
properties.put("leaderServes", "yes");
|
||||
properties.put("4lw.commands.whitelist", "*");
|
||||
properties.put("admin.enableServer", "false");
|
||||
QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig();
|
||||
quorumPeerConfig.parseProperties(properties);
|
||||
|
||||
ZooKeeperServerMain zookeeper = new ZooKeeperServerMain();
|
||||
final ServerConfig config = new ServerConfig();
|
||||
config.readFrom(quorumPeerConfig);
|
||||
Executors.newSingleThreadExecutor().submit(() -> {
|
||||
try {
|
||||
zookeeper.runFromConfig(config);
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
});
|
||||
log.info("Started TestZookeeper on port {}", port);
|
||||
return zookeeper;
|
||||
}
|
||||
|
||||
private int getNextPort(int start) {
|
||||
for (int port = start; port < start + MAX_PORT; ++port) {
|
||||
try {
|
||||
new ServerSocket(port).close();
|
||||
return port;
|
||||
} catch (IOException e) {
|
||||
|
||||
}
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
@@ -1,61 +0,0 @@
|
||||
package timely.balancer.test;
|
||||
|
||||
import org.apache.accumulo.core.client.AccumuloClient;
|
||||
import org.junit.rules.TestWatcher;
|
||||
import org.junit.runner.Description;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import timely.common.configuration.TimelyProperties;
|
||||
import timely.util.TestUtils;
|
||||
|
||||
public class BalancerTestRule extends TestWatcher {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(BalancerTestRule.class);
|
||||
private TimelyProperties timelyProperties;
|
||||
private AccumuloClient accumuloClient;
|
||||
|
||||
public BalancerTestRule(TimelyProperties timelyProperties, AccumuloClient accumuloClient) {
|
||||
this.timelyProperties = timelyProperties;
|
||||
this.accumuloClient = accumuloClient;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void failed(Throwable e, Description description) {
|
||||
log.error("Test {}:{} failed", description.getTestClass().getCanonicalName(), description.getMethodName());
|
||||
log.error(e.getMessage(), e);
|
||||
if (accumuloClient != null) {
|
||||
if (accumuloClient.tableOperations().exists(timelyProperties.getMetricsTable())) {
|
||||
separator();
|
||||
log.info("Contents of {}", timelyProperties.getMetricsTable());
|
||||
separator();
|
||||
TestUtils.printTable(accumuloClient, timelyProperties.getMetricsTable(), true);
|
||||
separator();
|
||||
}
|
||||
if (accumuloClient.tableOperations().exists(timelyProperties.getMetaTable())) {
|
||||
log.info("Contents of {}", timelyProperties.getMetaTable());
|
||||
separator();
|
||||
TestUtils.printTable(accumuloClient, timelyProperties.getMetaTable(), false);
|
||||
separator();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void separator() {
|
||||
log.info("-----------------------------------------------------------------------------------");
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void starting(Description description) {
|
||||
separator();
|
||||
log.info("Starting test {}:{}", description.getTestClass().getCanonicalName(), description.getMethodName());
|
||||
separator();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void finished(Description description) {
|
||||
separator();
|
||||
log.info("Finished test {}:{}", description.getTestClass().getCanonicalName(), description.getMethodName());
|
||||
separator();
|
||||
}
|
||||
}
|
||||
@@ -1,137 +0,0 @@
|
||||
package timely.balancer.test;
|
||||
|
||||
import static timely.Constants.SERVICE_DISCOVERY_PATH;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.state.ConnectionState;
|
||||
import org.apache.curator.x.discovery.ServiceCache;
|
||||
import org.apache.curator.x.discovery.ServiceDiscovery;
|
||||
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
|
||||
import org.apache.curator.x.discovery.ServiceInstance;
|
||||
import org.apache.curator.x.discovery.ServiceInstanceBuilder;
|
||||
import org.apache.curator.x.discovery.details.ServiceCacheListener;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.experimental.categories.Category;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
import timely.ServerDetails;
|
||||
import timely.balancer.component.TestHealthChecker;
|
||||
import timely.balancer.component.TestMetricResolver;
|
||||
import timely.balancer.connection.TimelyBalancedHost;
|
||||
import timely.test.IntegrationTest;
|
||||
|
||||
@Category(IntegrationTest.class)
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
public class MetricResolverIT {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(MetricResolverIT.class);
|
||||
|
||||
@Autowired
|
||||
@Rule
|
||||
public BalancerTestRule balancerTestRule;
|
||||
|
||||
@Autowired
|
||||
private TestMetricResolver metricResolver;
|
||||
|
||||
@Autowired
|
||||
private TestHealthChecker healthChecker;
|
||||
|
||||
@Autowired
|
||||
private CuratorFramework curatorFramework;
|
||||
|
||||
private ServiceDiscovery<ServerDetails> discovery;
|
||||
private ServiceCache<ServerDetails> serviceCache;
|
||||
|
||||
@Before
|
||||
public void setup() throws Exception {
|
||||
discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class).client(curatorFramework).basePath(SERVICE_DISCOVERY_PATH).build();
|
||||
discovery.start();
|
||||
serviceCache = discovery.serviceCacheBuilder().name("timely-server").build();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void ServiceRegistrationTest() {
|
||||
LinkedBlockingQueue<List<ServiceInstance<ServerDetails>>> updateQueue = getServiceUpdateQueue(serviceCache);
|
||||
registerService(curatorFramework, TimelyBalancedHost.of("host1", 5242, 5243, 5244, 5245, null));
|
||||
registerService(curatorFramework, TimelyBalancedHost.of("host1", 5342, 5343, 5344, 5345, null));
|
||||
registerService(curatorFramework, TimelyBalancedHost.of("host1", 5442, 5443, 5444, 5445, null));
|
||||
registerService(curatorFramework, TimelyBalancedHost.of("host1", 5542, 5543, 5544, 5545, null));
|
||||
try {
|
||||
List<ServiceInstance<ServerDetails>> currentUpdate;
|
||||
List<ServiceInstance<ServerDetails>> lastUpdate = null;
|
||||
do {
|
||||
currentUpdate = updateQueue.poll(200, TimeUnit.MILLISECONDS);
|
||||
if (currentUpdate != null) {
|
||||
lastUpdate = currentUpdate;
|
||||
}
|
||||
} while (currentUpdate != null);
|
||||
Assert.assertNotNull(lastUpdate);
|
||||
Assert.assertEquals(4, lastUpdate.size());
|
||||
} catch (InterruptedException e) {
|
||||
Assert.fail("" + e.getMessage());
|
||||
}
|
||||
Assert.assertEquals(4, metricResolver.getServerList().size());
|
||||
Assert.assertEquals(4, healthChecker.getTimelyHosts().size());
|
||||
}
|
||||
|
||||
public void registerService(CuratorFramework curatorFramework, TimelyBalancedHost server) {
|
||||
try {
|
||||
ServerDetails payload = new ServerDetails();
|
||||
payload.setHost(server.getHost());
|
||||
payload.setTcpPort(server.getTcpPort());
|
||||
payload.setHttpPort(server.getHttpPort());
|
||||
payload.setWsPort(server.getWsPort());
|
||||
payload.setUdpPort(server.getUdpPort());
|
||||
|
||||
ServiceInstanceBuilder<ServerDetails> builder = ServiceInstance.builder();
|
||||
String serviceName = server.getHost() + ":" + server.getTcpPort();
|
||||
ServiceInstance<ServerDetails> serviceInstance = builder.id(serviceName).name("timely-server").address(server.getHost()).port(server.getTcpPort())
|
||||
.payload(payload).build();
|
||||
|
||||
ServiceDiscovery<ServerDetails> discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class).client(curatorFramework)
|
||||
.basePath(SERVICE_DISCOVERY_PATH).build();
|
||||
discovery.start();
|
||||
discovery.registerService(serviceInstance);
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private LinkedBlockingQueue<List<ServiceInstance<ServerDetails>>> getServiceUpdateQueue(ServiceCache<ServerDetails> serviceCache) {
|
||||
|
||||
LinkedBlockingQueue<List<ServiceInstance<ServerDetails>>> updates = new LinkedBlockingQueue<>();
|
||||
try {
|
||||
|
||||
ServiceCacheListener listener = new ServiceCacheListener() {
|
||||
|
||||
@Override
|
||||
public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cacheChanged() {
|
||||
updates.add(serviceCache.getInstances());
|
||||
}
|
||||
};
|
||||
serviceCache.addListener(listener);
|
||||
serviceCache.start();
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
return updates;
|
||||
}
|
||||
}
|
||||
@@ -1,48 +0,0 @@
|
||||
package timely.balancer.test;
|
||||
|
||||
public class TestServerInfo {
|
||||
|
||||
public String host;
|
||||
public String ip;
|
||||
public Integer tcpPort;
|
||||
public Integer httpPort;
|
||||
public Integer wsPort;
|
||||
public Integer udpPort;
|
||||
|
||||
public TestServerInfo(String host, String ip, Integer tcpPort, Integer httpPort, Integer wsPort, Integer udpPort) {
|
||||
this.host = host;
|
||||
this.ip = ip;
|
||||
this.tcpPort = tcpPort;
|
||||
this.httpPort = httpPort;
|
||||
this.wsPort = wsPort;
|
||||
this.udpPort = udpPort;
|
||||
}
|
||||
|
||||
static public TestServerInfo of(String host, String ip, Integer tcpPort, Integer httpPort, Integer wsPort, Integer udpPort) {
|
||||
return new TestServerInfo(host, ip, tcpPort, httpPort, wsPort, udpPort);
|
||||
}
|
||||
|
||||
public String getHost() {
|
||||
return host;
|
||||
}
|
||||
|
||||
public String getIp() {
|
||||
return ip;
|
||||
}
|
||||
|
||||
public Integer getTcpPort() {
|
||||
return tcpPort;
|
||||
}
|
||||
|
||||
public Integer getHttpPort() {
|
||||
return httpPort;
|
||||
}
|
||||
|
||||
public Integer getWsPort() {
|
||||
return wsPort;
|
||||
}
|
||||
|
||||
public Integer getUdpPort() {
|
||||
return udpPort;
|
||||
}
|
||||
}
|
||||
@@ -1,141 +0,0 @@
|
||||
logging:
|
||||
level:
|
||||
ROOT: info
|
||||
timely: debug
|
||||
org.apache.zookeeper: error
|
||||
|
||||
timely:
|
||||
test: true
|
||||
mini-accumulo:
|
||||
enabled: true
|
||||
rootAuths: A,B,C,D,E,F,G,H,I
|
||||
zookeeper:
|
||||
servers: localhost:0
|
||||
timeout: 30s
|
||||
security:
|
||||
session-max-age: 86400
|
||||
cache-expiration-minutes: 60
|
||||
cache-refresh-minutes: 5
|
||||
allow-anonymous-http-access: true
|
||||
allow-anonymous-ws-access: true
|
||||
authorizedUsers:
|
||||
- subjectDn: cn=example.com
|
||||
roles: G,H,I
|
||||
auths: D,E,F
|
||||
userType: USER
|
||||
- subjectDn: cn=test a. user, ou=example developers, o=example corp, c=us
|
||||
issuerDn: cn=example corp ca, o=example corp, c=us
|
||||
roles: G,H,I
|
||||
auths: A,B,C
|
||||
userType: USER
|
||||
serverSsl:
|
||||
useGeneratedKeypair: true
|
||||
ignoreSslHandshakeErrors: true
|
||||
useOpenssl: true
|
||||
cache:
|
||||
enabled: false
|
||||
metric-age-off-hours:
|
||||
default: 26
|
||||
nonCachedMetrics:
|
||||
balancer:
|
||||
checkServerHealthInterval: 10000
|
||||
serverFailuresBeforeDown: 6
|
||||
serverSuccessesBeforeUp: 1
|
||||
assignmentFile: file://${java.io.tmpdir}/timely/balancer/assignments.csv
|
||||
server:
|
||||
ip: 127.0.0.1
|
||||
tcpPort: 4242
|
||||
numTcpPools: 1
|
||||
tcpBufferSize: 100
|
||||
tcpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
udpPort: 4245
|
||||
udpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
http:
|
||||
host: localhost
|
||||
ip: 127.0.0.1
|
||||
port: 4243
|
||||
redirect-path: /secure-me
|
||||
strict-transport-max-age: 604800
|
||||
cors:
|
||||
allow-any-origin: false
|
||||
allow-null-origin: false
|
||||
allow-credentials: true
|
||||
allowed-origins:
|
||||
allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
|
||||
allowed-headers: content-type
|
||||
httpClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
websocket:
|
||||
ip: 127.0.0.1
|
||||
port: 5244
|
||||
timeout: 60
|
||||
subscription-lag: 120
|
||||
scanner-batch-size: 10000
|
||||
flush-interval-seconds: 30
|
||||
scanner-read-ahead: 1
|
||||
subscription-batch-size: 5000
|
||||
incomingBufferSize: 50000000
|
||||
wsClientPool:
|
||||
minIdlePerKey: 0
|
||||
maxIdlePerKey: 40
|
||||
maxTotalPerKey: 40
|
||||
maxTotal: -1
|
||||
lifo: true
|
||||
fairness: false
|
||||
maxWaitMillis: -1
|
||||
minEvictableIdleTimeMillis: 1800000
|
||||
evictorShutdownTimeoutMillis: 10000
|
||||
softMinEvictableIdleTimeMillis: -1
|
||||
numTestsPerEvictionRun: 3
|
||||
timeBetweenEvictionRunsMillis: -1
|
||||
blockWhenExhausted: true
|
||||
jmxEnabled: true
|
||||
testOnBorrow: false
|
||||
testOnReturn: false
|
||||
@@ -1,3 +0,0 @@
|
||||
spring:
|
||||
application:
|
||||
name: timely-balancer
|
||||
@@ -1,24 +0,0 @@
|
||||
Configuration:
|
||||
status: warn
|
||||
monitorInterval: 60
|
||||
|
||||
Properties:
|
||||
Property:
|
||||
- name: PID
|
||||
value: "????"
|
||||
- name: LOG_PATTERN
|
||||
value: "%clr{%d{yyyy-MM-dd HH:mm:ss.SSS}}{faint} %clr{%5p} %clr{${sys:PID}}{magenta} %clr{---}{faint} %clr{[%15.15t]}{faint} %clr{%-40.40c{1.}}{cyan} %clr{:}{faint} %m%n%wEx"
|
||||
|
||||
Appenders:
|
||||
Console:
|
||||
name: Console
|
||||
target: SYSTEM_OUT
|
||||
follow: true
|
||||
PatternLayout:
|
||||
pattern: "${LOG_PATTERN}"
|
||||
|
||||
Loggers:
|
||||
Root:
|
||||
level: warn
|
||||
AppenderRef:
|
||||
- ref: Console
|
||||
+198
-147
@@ -1,151 +1,202 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<?xml version="1.0"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely</artifactId>
|
||||
<version>2.1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>timely-client</artifactId>
|
||||
<properties>
|
||||
<!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
|
||||
<!-- use .dev0 instead of -SNAPSHOT -->
|
||||
<python.client.version>2.1.0.dev0</python.client.version>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-annotations</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.datatype</groupId>
|
||||
<artifactId>jackson-datatype-jdk8</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.google.guava</groupId>
|
||||
<artifactId>guava</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-lang3</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.httpcomponents</groupId>
|
||||
<artifactId>httpclient</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.glassfish.tyrus</groupId>
|
||||
<artifactId>tyrus-client</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.glassfish.tyrus</groupId>
|
||||
<artifactId>tyrus-container-grizzly-client</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-common</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-api</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-core</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<!--
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-slf4j-impl</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
-->
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-resources-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>filter-setup</id>
|
||||
<goals>
|
||||
<goal>copy-resources</goal>
|
||||
</goals>
|
||||
<phase>process-resources</phase>
|
||||
<configuration>
|
||||
<outputDirectory>${project.basedir}/src/main/python</outputDirectory>
|
||||
<resources>
|
||||
<resource>
|
||||
<directory>${project.basedir}/src/main/python/template</directory>
|
||||
<includes>
|
||||
<include>setup.py</include>
|
||||
</includes>
|
||||
<filtering>true</filtering>
|
||||
</resource>
|
||||
</resources>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
<profiles>
|
||||
<profile>
|
||||
<id>python</id>
|
||||
<build>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>timely-client</artifactId>
|
||||
<parent>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely</artifactId>
|
||||
<version>0.0.6-SNAPSHOT</version>
|
||||
</parent>
|
||||
<properties>
|
||||
<!-- PEP 440 compliant version gets supstituted into src/main/python/template/setup.py -->
|
||||
<python.client.version>0.0.6.dev0</python.client.version>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>timely</groupId>
|
||||
<artifactId>timely-common</artifactId>
|
||||
<version>${parent.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-annotations</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.datatype</groupId>
|
||||
<artifactId>jackson-datatype-jdk8</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.google.guava</groupId>
|
||||
<artifactId>guava</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-lang3</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.httpcomponents</groupId>
|
||||
<artifactId>httpclient</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-api</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-core</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-slf4j-impl</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.glassfish.tyrus</groupId>
|
||||
<artifactId>tyrus-client</artifactId>
|
||||
<version>1.13</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.glassfish.tyrus</groupId>
|
||||
<artifactId>tyrus-container-grizzly-client</artifactId>
|
||||
<version>1.13</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>exec-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>python-dist</id>
|
||||
<goals>
|
||||
<goal>exec</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<executable>python</executable>
|
||||
<workingDirectory>${project.basedir}/src/main/python</workingDirectory>
|
||||
<commandlineArgs>setup.py sdist --dist-dir ${project.build.directory}</commandlineArgs>
|
||||
</configuration>
|
||||
</execution>
|
||||
<execution>
|
||||
<id>python-clean</id>
|
||||
<goals>
|
||||
<goal>exec</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<executable>rm</executable>
|
||||
<workingDirectory>${project.basedir}/src/main/python</workingDirectory>
|
||||
<commandlineArgs>-rf timely.egg-info setup.py</commandlineArgs>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>findbugs-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-jar-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.jacoco</groupId>
|
||||
<artifactId>jacoco-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>net.revelc.code</groupId>
|
||||
<artifactId>formatter-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>buildnumber-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<artifactId>maven-resources-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>filter-setup</id>
|
||||
<goals>
|
||||
<goal>copy-resources</goal>
|
||||
</goals>
|
||||
<phase>process-resources</phase>
|
||||
<configuration>
|
||||
<outputDirectory>${project.basedir}/src/main/python</outputDirectory>
|
||||
<resources>
|
||||
<resource>
|
||||
<directory>${project.basedir}/src/main/python/template</directory>
|
||||
<includes>
|
||||
<include>setup.py</include>
|
||||
</includes>
|
||||
<filtering>true</filtering>
|
||||
</resource>
|
||||
</resources>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<!--
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>exec-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>python-dist</id>
|
||||
<goals>
|
||||
<goal>exec</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<executable>python</executable>
|
||||
<workingDirectory>${project.basedir}/src/main/python</workingDirectory>
|
||||
<commandlineArgs>setup.py sdist - -dist-dir ${project.build.directory}</commandlineArgs>
|
||||
</configuration>
|
||||
</execution>
|
||||
<execution>
|
||||
<id>python-clean</id>
|
||||
<goals>
|
||||
<goal>exec</goal>
|
||||
</goals>
|
||||
<phase>package</phase>
|
||||
<configuration>
|
||||
<executable>rm</executable>
|
||||
<workingDirectory>${project.basedir}/src/main/python</workingDirectory>
|
||||
<commandlineArgs>-rf timely.egg-info setup.py</commandlineArgs>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
-->
|
||||
</plugins>
|
||||
</build>
|
||||
</profile>
|
||||
</profiles>
|
||||
</build>
|
||||
<reporting>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>findbugs-maven-plugin</artifactId>
|
||||
<version>${version.findbugs-maven}</version>
|
||||
<configuration>
|
||||
<effort>Max</effort>
|
||||
<threshold>Medium</threshold>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.jacoco</groupId>
|
||||
<artifactId>jacoco-maven-plugin</artifactId>
|
||||
<version>${version.jacoco}</version>
|
||||
<reportSets>
|
||||
<reportSet>
|
||||
<id>JaCoCo Unit Test Coverage</id>
|
||||
<configuration>
|
||||
<dataFile>${project.build.directory}/coverage/jacoco-ut.exec</dataFile>
|
||||
</configuration>
|
||||
<reports>
|
||||
<report>report</report>
|
||||
</reports>
|
||||
</reportSet>
|
||||
<reportSet>
|
||||
<id>JaCoCo Integration Test Coverage</id>
|
||||
<configuration>
|
||||
<dataFile>${project.build.directory}/coverage/jacoco-it.exec</dataFile>
|
||||
</configuration>
|
||||
<reports>
|
||||
<report>report-integration</report>
|
||||
</reports>
|
||||
</reportSet>
|
||||
</reportSets>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</reporting>
|
||||
</project>
|
||||
@@ -7,7 +7,7 @@ public class AddSubscription extends SubscriptionRequest {
|
||||
|
||||
private String operation = "add";
|
||||
private String metric = null;
|
||||
private Optional<Map<String,String>> tags = Optional.empty();
|
||||
private Optional<Map<String, String>> tags = Optional.empty();
|
||||
private Optional<Long> startTime = Optional.empty();
|
||||
private Optional<Long> endTime = Optional.empty();
|
||||
private Optional<Long> delayTime = Optional.empty();
|
||||
@@ -20,7 +20,7 @@ public class AddSubscription extends SubscriptionRequest {
|
||||
return metric;
|
||||
}
|
||||
|
||||
public Optional<Map<String,String>> getTags() {
|
||||
public Optional<Map<String, String>> getTags() {
|
||||
return tags;
|
||||
}
|
||||
|
||||
@@ -36,7 +36,7 @@ public class AddSubscription extends SubscriptionRequest {
|
||||
this.metric = metric;
|
||||
}
|
||||
|
||||
public void setTags(Optional<Map<String,String>> tags) {
|
||||
public void setTags(Optional<Map<String, String>> tags) {
|
||||
this.tags = tags;
|
||||
}
|
||||
|
||||
@@ -71,8 +71,9 @@ public class AddSubscription extends SubscriptionRequest {
|
||||
}
|
||||
if (obj instanceof AddSubscription) {
|
||||
AddSubscription other = (AddSubscription) obj;
|
||||
return (this.subscriptionId.equals(other.subscriptionId) && this.metric.equals(other.metric) && this.tags.equals(other.tags)
|
||||
&& this.startTime.equals(other.startTime) && this.endTime.equals(other.endTime) && this.delayTime.equals(other.delayTime));
|
||||
return (this.subscriptionId.equals(other.subscriptionId) && this.metric.equals(other.metric)
|
||||
&& this.tags.equals(other.tags) && this.startTime.equals(other.startTime)
|
||||
&& this.endTime.equals(other.endTime) && this.delayTime.equals(other.delayTime));
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package timely.api;
|
||||
|
||||
public class BasicAuthLogin {
|
||||
|
||||
protected String username;
|
||||
protected String password;
|
||||
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
public void setUsername(String user) {
|
||||
this.username = user;
|
||||
}
|
||||
|
||||
public String getPassword() {
|
||||
return password;
|
||||
}
|
||||
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return (username + password).hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (null == obj) {
|
||||
return false;
|
||||
}
|
||||
if (obj == this) {
|
||||
return true;
|
||||
}
|
||||
if (obj instanceof BasicAuthLogin) {
|
||||
BasicAuthLogin other = (BasicAuthLogin) obj;
|
||||
return (this.username.equals(other.username) && this.password.equals(other.password));
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,10 +0,0 @@
|
||||
package timely.api;
|
||||
|
||||
public class VersionRequest {
|
||||
|
||||
private String operation = "version";
|
||||
|
||||
public String getOperation() {
|
||||
return operation;
|
||||
}
|
||||
}
|
||||
@@ -6,12 +6,9 @@ import org.apache.commons.lang3.builder.EqualsBuilder;
|
||||
import org.apache.commons.lang3.builder.HashCodeBuilder;
|
||||
import org.apache.commons.lang3.builder.ToStringBuilder;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
|
||||
|
||||
import timely.model.Metric;
|
||||
import timely.model.Tag;
|
||||
|
||||
@JsonPropertyOrder({"metric", "timestamp", "value", "tags", "subscriptionId", "complete"})
|
||||
public class MetricResponse {
|
||||
|
||||
private String metric;
|
||||
|
||||
@@ -1,13 +1,12 @@
|
||||
package timely.client.http;
|
||||
|
||||
import javax.net.ssl.HostnameVerifier;
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import org.apache.http.client.CookieStore;
|
||||
import org.apache.http.client.config.CookieSpecs;
|
||||
import org.apache.http.client.config.RequestConfig;
|
||||
import org.apache.http.conn.ssl.DefaultHostnameVerifier;
|
||||
import org.apache.http.conn.ssl.NoopHostnameVerifier;
|
||||
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.apache.http.impl.client.HttpClientBuilder;
|
||||
import org.apache.http.impl.client.HttpClients;
|
||||
@@ -15,8 +14,8 @@ import org.glassfish.tyrus.client.SslContextConfigurator;
|
||||
|
||||
public class HttpClient {
|
||||
|
||||
public static SSLContext getSSLContext(String trustStoreFile, String trustStoreType, String trustStorePass, String keyStoreFile, String keyStoreType,
|
||||
String keyStorePass) {
|
||||
public static SSLContext getSSLContext(String trustStoreFile, String trustStoreType, String trustStorePass,
|
||||
String keyStoreFile, String keyStoreType, String keyStorePass) {
|
||||
SslContextConfigurator scc = new SslContextConfigurator();
|
||||
scc.setTrustStoreFile(trustStoreFile);
|
||||
scc.setTrustStoreType(trustStoreType);
|
||||
@@ -31,37 +30,24 @@ public class HttpClient {
|
||||
return scc.createSSLContext();
|
||||
}
|
||||
|
||||
public static CloseableHttpClient get(String trustStoreFile, String trustStoreType, String trustStorePass, String keyStoreFile, String keyStoreType,
|
||||
String keyStorePass, CookieStore cookieStore, boolean hostVerificationEnabled) {
|
||||
public static CloseableHttpClient get(String trustStoreFile, String trustStoreType, String trustStorePass,
|
||||
String keyStoreFile, String keyStoreType, String keyStorePass, CookieStore cookieStore,
|
||||
boolean hostVerificationEnabled) {
|
||||
|
||||
SSLContext ssl = getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType, keyStorePass);
|
||||
SSLContext ssl = getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType,
|
||||
keyStorePass);
|
||||
return get(ssl, cookieStore, hostVerificationEnabled);
|
||||
}
|
||||
|
||||
public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled) {
|
||||
return get(ssl, cookieStore, hostVerificationEnabled, false);
|
||||
}
|
||||
|
||||
public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled, boolean clientAuth) {
|
||||
HostnameVerifier hostnameVerifier;
|
||||
if (hostVerificationEnabled) {
|
||||
hostnameVerifier = SSLConnectionSocketFactory.getDefaultHostnameVerifier();
|
||||
} else {
|
||||
hostnameVerifier = new NoopHostnameVerifier();
|
||||
}
|
||||
RequestConfig defaultRequestConfig = RequestConfig.custom().setCookieSpec(CookieSpecs.STANDARD).build();
|
||||
HttpClientBuilder builder;
|
||||
if (cookieStore == null) {
|
||||
builder = HttpClients.custom().setSSLContext(ssl).disableCookieManagement().setDefaultRequestConfig(defaultRequestConfig)
|
||||
.setSSLHostnameVerifier(hostnameVerifier);
|
||||
} else {
|
||||
builder = HttpClients.custom().setSSLContext(ssl).setDefaultCookieStore(cookieStore).setDefaultRequestConfig(defaultRequestConfig)
|
||||
.setSSLHostnameVerifier(hostnameVerifier);
|
||||
}
|
||||
|
||||
if (clientAuth) {
|
||||
SSLConnectionSocketFactory sslCSF = new SSLConnectionSocketFactory(ssl, new String[] {"TLSv1.1", "TLSv1.2"}, null, hostnameVerifier);
|
||||
builder.setSSLSocketFactory(sslCSF);
|
||||
HttpClientBuilder builder = HttpClients.custom().setSSLContext(ssl).setDefaultCookieStore(cookieStore)
|
||||
.setDefaultRequestConfig(defaultRequestConfig);
|
||||
if (hostVerificationEnabled) {
|
||||
builder.setSSLHostnameVerifier(new DefaultHostnameVerifier());
|
||||
} else {
|
||||
builder.setSSLHostnameVerifier(new NoopHostnameVerifier());
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
@@ -1,40 +1,31 @@
|
||||
package timely.client.tcp;
|
||||
|
||||
import java.io.BufferedWriter;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStreamWriter;
|
||||
import java.io.PrintWriter;
|
||||
import java.net.Socket;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
public class TcpClient implements AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(TcpClient.class);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TcpClient.class);
|
||||
|
||||
private final String host;
|
||||
private final int port;
|
||||
private Socket sock = null;
|
||||
private BufferedWriter out = null;
|
||||
private PrintWriter out = null;
|
||||
private long connectTime = 0L;
|
||||
private long backoff = 2000;
|
||||
private int bufferSize = -1;
|
||||
private int writesSinceFlush = 0;
|
||||
|
||||
public TcpClient(String hostname, int port) {
|
||||
this(hostname, port, -1);
|
||||
}
|
||||
|
||||
public TcpClient(String hostname, int port, int bufferSize) {
|
||||
this.host = hostname;
|
||||
this.port = port;
|
||||
this.bufferSize = bufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* Opens a TCP connection to the specified host and port
|
||||
*
|
||||
*
|
||||
* @throws IOException
|
||||
* if an error occurs
|
||||
*/
|
||||
@@ -46,48 +37,41 @@ public class TcpClient implements AutoCloseable {
|
||||
|
||||
/**
|
||||
* Write a metric to Timely
|
||||
*
|
||||
*
|
||||
* @param metric
|
||||
* newline terminated string representation of Timely metric
|
||||
* @throws IOException
|
||||
*/
|
||||
public synchronized void write(String metric) throws IOException {
|
||||
if (connect() != 0) {
|
||||
throw new IOException();
|
||||
}
|
||||
out.write(metric);
|
||||
writesSinceFlush++;
|
||||
if (bufferSize > 0 && writesSinceFlush >= bufferSize) {
|
||||
flush();
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void flush() throws IOException {
|
||||
public synchronized void flush() {
|
||||
if (null != out) {
|
||||
out.flush();
|
||||
writesSinceFlush = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes the tcp connection to Timely
|
||||
*
|
||||
*
|
||||
* @throws IOException
|
||||
* if an error occurs
|
||||
*/
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
log.trace("Shutting down connection to Timely at {}:{}", host, port);
|
||||
LOG.info("Shutting down connection to Timely at {}:{}", host, port);
|
||||
if (null != sock) {
|
||||
try {
|
||||
if (null != out) {
|
||||
if (bufferSize > 0) {
|
||||
flush();
|
||||
}
|
||||
out.close();
|
||||
}
|
||||
sock.close();
|
||||
} catch (IOException e) {
|
||||
log.error("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
|
||||
LOG.error("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
|
||||
} finally {
|
||||
sock = null;
|
||||
}
|
||||
@@ -95,56 +79,28 @@ public class TcpClient implements AutoCloseable {
|
||||
}
|
||||
|
||||
private synchronized int connect() {
|
||||
if (null == sock || !sock.isConnected()) {
|
||||
if (null == sock || !sock.isConnected() || out.checkError()) {
|
||||
if (System.currentTimeMillis() > (connectTime + backoff)) {
|
||||
OutputStreamWriter osw = null;
|
||||
try {
|
||||
connectTime = System.currentTimeMillis();
|
||||
sock = new Socket(host, port);
|
||||
osw = new OutputStreamWriter(sock.getOutputStream(), StandardCharsets.UTF_8);
|
||||
out = new BufferedWriter(osw);
|
||||
out = new PrintWriter(sock.getOutputStream(), false);
|
||||
backoff = 2000;
|
||||
log.trace("Connected to Timely at {}:{}", host, port);
|
||||
} catch (Exception e) {
|
||||
log.error("Error connecting to Timely at {}:{} - {}", host, port, e.getMessage());
|
||||
if (sock != null) {
|
||||
try {
|
||||
sock.close();
|
||||
} catch (IOException e1) {
|
||||
log.error(e1.getMessage());
|
||||
} finally {
|
||||
sock = null;
|
||||
}
|
||||
}
|
||||
if (osw != null) {
|
||||
try {
|
||||
osw.close();
|
||||
} catch (IOException e1) {
|
||||
log.error(e1.getMessage());
|
||||
}
|
||||
}
|
||||
if (out != null) {
|
||||
try {
|
||||
out.close();
|
||||
} catch (IOException e1) {
|
||||
log.error(e1.getMessage());
|
||||
} finally {
|
||||
out = null;
|
||||
}
|
||||
}
|
||||
LOG.info("Connected to Timely at {}:{}", host, port);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Error connecting to Timely at {}:" + host + ":" + port + ". Error: " + e.getMessage());
|
||||
backoff = backoff * 2;
|
||||
log.info("Will retry connection in {} ms.", backoff);
|
||||
sock = null;
|
||||
out = null;
|
||||
LOG.warn("Will retry connection in {} ms.", backoff);
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
log.warn("Not writing to Timely, waiting to reconnect");
|
||||
LOG.warn("Not writing to Timely, waiting to reconnect");
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
public long getConnectTime() {
|
||||
return connectTime;
|
||||
}
|
||||
}
|
||||
@@ -7,24 +7,15 @@ import java.net.DatagramPacket;
|
||||
import java.net.DatagramSocket;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
public class UdpClient implements AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(UdpClient.class);
|
||||
|
||||
private final String host;
|
||||
private final int port;
|
||||
private final InetSocketAddress address;
|
||||
private final DatagramPacket packet;
|
||||
private DatagramSocket sock;
|
||||
|
||||
public UdpClient(String host, int port) {
|
||||
this.address = new InetSocketAddress(host, port);
|
||||
public UdpClient(String hostname, int port) {
|
||||
this.address = new InetSocketAddress(hostname, port);
|
||||
this.packet = new DatagramPacket("".getBytes(UTF_8), 0, 0, address.getAddress(), port);
|
||||
this.host = host;
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public void open() throws IOException {
|
||||
@@ -38,11 +29,11 @@ public class UdpClient implements AutoCloseable {
|
||||
throw new IllegalStateException("Must call open first");
|
||||
}
|
||||
this.packet.setData(metric.getBytes(UTF_8));
|
||||
log.info("writing '" + metric + "' to " + this.host + ":" + this.port);
|
||||
this.sock.send(packet);
|
||||
}
|
||||
|
||||
public void flush() throws IOException {}
|
||||
public void flush() throws IOException {
|
||||
}
|
||||
|
||||
public void close() throws IOException {
|
||||
if (null != sock) {
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
package timely.client.websocket;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import javax.websocket.CloseReason;
|
||||
import javax.websocket.Endpoint;
|
||||
import javax.websocket.EndpointConfig;
|
||||
import javax.websocket.MessageHandler;
|
||||
import javax.websocket.Session;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
@@ -13,25 +11,28 @@ import org.slf4j.LoggerFactory;
|
||||
|
||||
public abstract class ClientHandler extends Endpoint {
|
||||
|
||||
private final static Logger log = LoggerFactory.getLogger(ClientHandler.class);
|
||||
private final static Logger LOG = LoggerFactory.getLogger(ClientHandler.class);
|
||||
|
||||
@Override
|
||||
public void onOpen(Session session, EndpointConfig config) {
|
||||
log.info("Websocket session {} opened.", session.getId());
|
||||
session.addMessageHandler(String.class, message -> {
|
||||
log.info("Message received on Websocket session {}: {}", session.getId(), message);
|
||||
LOG.info("Websocket session {} opened.", session.getId());
|
||||
session.addMessageHandler(new MessageHandler.Whole<String>() {
|
||||
|
||||
@Override
|
||||
public void onMessage(String message) {
|
||||
LOG.info("Message received on Websocket session {}: {}", session.getId(), message);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onClose(Session session, CloseReason reason) {
|
||||
log.info("Websocket session {} closed.", session.getId());
|
||||
LOG.info("Websocket session {} closed.", session.getId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Session session, Throwable error) {
|
||||
log.error("Error occurred on Websocket session" + session.getId(), error);
|
||||
LOG.error("Error occurred on Websocket session" + session.getId(), error);
|
||||
}
|
||||
|
||||
public void beforeRequest(Map<String,List<String>> headers) {}
|
||||
}
|
||||
@@ -14,15 +14,13 @@ import org.apache.http.cookie.Cookie;
|
||||
public class TimelyEndpointConfig implements ClientEndpointConfig {
|
||||
|
||||
private String sessionCookie = null;
|
||||
private final ClientHandler clientEndpoint;
|
||||
|
||||
private static final String FORMAT = "%s=%s;";
|
||||
|
||||
public TimelyEndpointConfig(ClientHandler clientEndpoint, Cookie sessionCookie) {
|
||||
public TimelyEndpointConfig(Cookie sessionCookie) {
|
||||
if (null != sessionCookie) {
|
||||
this.sessionCookie = String.format(FORMAT, sessionCookie.getName(), sessionCookie.getValue());
|
||||
}
|
||||
this.clientEndpoint = clientEndpoint;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -36,7 +34,7 @@ public class TimelyEndpointConfig implements ClientEndpointConfig {
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String,Object> getUserProperties() {
|
||||
public Map<String, Object> getUserProperties() {
|
||||
return Collections.emptyMap();
|
||||
}
|
||||
|
||||
@@ -55,12 +53,11 @@ public class TimelyEndpointConfig implements ClientEndpointConfig {
|
||||
return new Configurator() {
|
||||
|
||||
@Override
|
||||
public void beforeRequest(Map<String,List<String>> headers) {
|
||||
public void beforeRequest(Map<String, List<String>> headers) {
|
||||
super.beforeRequest(headers);
|
||||
if (null != sessionCookie) {
|
||||
headers.put("Cookie", Collections.singletonList(sessionCookie));
|
||||
}
|
||||
TimelyEndpointConfig.this.clientEndpoint.beforeRequest(headers);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@@ -10,13 +10,16 @@ import javax.net.ssl.SSLContext;
|
||||
import javax.websocket.DeploymentException;
|
||||
import javax.websocket.Session;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.apache.http.HttpResponse;
|
||||
import org.apache.http.HttpStatus;
|
||||
import org.apache.http.client.HttpResponseException;
|
||||
import org.apache.http.client.methods.HttpGet;
|
||||
import org.apache.http.client.methods.HttpPost;
|
||||
import org.apache.http.client.methods.HttpRequestBase;
|
||||
import org.apache.http.client.utils.HttpClientUtils;
|
||||
import org.apache.http.cookie.Cookie;
|
||||
import org.apache.http.entity.StringEntity;
|
||||
import org.apache.http.impl.client.BasicCookieStore;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.glassfish.tyrus.client.ClientManager;
|
||||
@@ -25,19 +28,22 @@ import org.glassfish.tyrus.client.SslEngineConfigurator;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
import timely.api.BasicAuthLogin;
|
||||
import timely.client.http.HttpClient;
|
||||
import timely.serialize.JsonSerializer;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
public class WebSocketClient implements AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(WebSocketClient.class);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(WebSocketClient.class);
|
||||
|
||||
private final String timelyHostname;
|
||||
private final int timelyHttpsPort;
|
||||
private final int timelyWssPort;
|
||||
private final boolean doLogin;
|
||||
private final boolean clientAuth;
|
||||
private final String timelyUsername;
|
||||
private final String timelyPassword;
|
||||
private final boolean hostVerificationEnabled;
|
||||
private final int bufferSize;
|
||||
private final SSLContext ssl;
|
||||
@@ -46,46 +52,72 @@ public class WebSocketClient implements AutoCloseable {
|
||||
protected Session session = null;
|
||||
protected volatile boolean closed = true;
|
||||
|
||||
protected WebSocketClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin,
|
||||
boolean hostVerificationEnabled, int bufferSize) {
|
||||
protected WebSocketClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort,
|
||||
boolean doLogin, String timelyUsername, String timelyPassword, boolean hostVerificationEnabled,
|
||||
int bufferSize) {
|
||||
this.ssl = ssl;
|
||||
this.timelyHostname = timelyHostname;
|
||||
this.timelyHttpsPort = timelyHttpsPort;
|
||||
this.timelyWssPort = timelyWssPort;
|
||||
this.doLogin = doLogin;
|
||||
this.clientAuth = clientAuth;
|
||||
this.timelyUsername = timelyUsername;
|
||||
this.timelyPassword = timelyPassword;
|
||||
this.hostVerificationEnabled = hostVerificationEnabled;
|
||||
this.bufferSize = bufferSize;
|
||||
|
||||
Preconditions.checkNotNull(timelyHostname, "%s must be supplied", "Timely host name");
|
||||
Preconditions.checkNotNull(timelyHttpsPort, "%s must be supplied", "Timely HTTPS port");
|
||||
Preconditions.checkNotNull(timelyWssPort, "%s must be supplied", "Timely WSS port");
|
||||
|
||||
if (doLogin
|
||||
&& ((StringUtils.isEmpty(timelyUsername) && !StringUtils.isEmpty(timelyPassword) || (!StringUtils
|
||||
.isEmpty(timelyUsername) && StringUtils.isEmpty(timelyPassword))))) {
|
||||
throw new IllegalArgumentException("Both Timely username and password must be empty or non-empty");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
protected WebSocketClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String keyStoreFile,
|
||||
String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
|
||||
boolean hostVerificationEnabled, int bufferSize) {
|
||||
this(HttpClient.getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType, keyStorePass), timelyHostname,
|
||||
timelyHttpsPort, timelyWssPort, clientAuth, doLogin, hostVerificationEnabled, bufferSize);
|
||||
protected WebSocketClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean doLogin,
|
||||
String timelyUsername, String timelyPassword, String keyStoreFile, String keyStoreType,
|
||||
String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
|
||||
boolean hostVerificationEnabled, int bufferSize) {
|
||||
this(HttpClient.getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType,
|
||||
keyStorePass), timelyHostname, timelyHttpsPort, timelyWssPort, doLogin, timelyUsername, timelyPassword,
|
||||
hostVerificationEnabled, bufferSize);
|
||||
}
|
||||
|
||||
public void open(ClientHandler clientEndpoint) throws IOException, DeploymentException, URISyntaxException {
|
||||
|
||||
Cookie sessionCookie = null;
|
||||
if (doLogin) {
|
||||
BasicCookieStore cookieJar = new BasicCookieStore();
|
||||
try (CloseableHttpClient client = HttpClient.get(ssl, cookieJar, hostVerificationEnabled, clientAuth)) {
|
||||
try (CloseableHttpClient client = HttpClient.get(ssl, cookieJar, hostVerificationEnabled)) {
|
||||
|
||||
String target = "https://" + timelyHostname + ":" + timelyHttpsPort + "/login";
|
||||
|
||||
HttpRequestBase request = null;
|
||||
// HTTP GET to /login to use certificate based login
|
||||
request = new HttpGet(target);
|
||||
log.trace("Performing client certificate login");
|
||||
if (StringUtils.isEmpty(timelyUsername)) {
|
||||
// HTTP GET to /login to use certificate based login
|
||||
request = new HttpGet(target);
|
||||
LOG.trace("Performing client certificate login");
|
||||
} else {
|
||||
// HTTP POST to /login to use username/password
|
||||
BasicAuthLogin login = new BasicAuthLogin();
|
||||
login.setUsername(timelyUsername);
|
||||
login.setPassword(timelyPassword);
|
||||
String payload = JsonSerializer.getObjectMapper().writeValueAsString(login);
|
||||
HttpPost post = new HttpPost(target);
|
||||
post.setEntity(new StringEntity(payload));
|
||||
request = post;
|
||||
LOG.trace("Performing BasicAuth login");
|
||||
}
|
||||
|
||||
HttpResponse response = null;
|
||||
try {
|
||||
response = client.execute(request);
|
||||
if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
|
||||
throw new HttpResponseException(response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase());
|
||||
throw new HttpResponseException(response.getStatusLine().getStatusCode(), response
|
||||
.getStatusLine().getReasonPhrase());
|
||||
}
|
||||
for (Cookie c : cookieJar.getCookies()) {
|
||||
if (c.getName().equals("TSESSIONID")) {
|
||||
@@ -94,7 +126,8 @@ public class WebSocketClient implements AutoCloseable {
|
||||
}
|
||||
}
|
||||
if (null == sessionCookie) {
|
||||
throw new IllegalStateException("Unable to find TSESSIONID cookie header in Timely login response");
|
||||
throw new IllegalStateException(
|
||||
"Unable to find TSESSIONID cookie header in Timely login response");
|
||||
}
|
||||
} finally {
|
||||
HttpClientUtils.closeQuietly(response);
|
||||
@@ -110,14 +143,15 @@ public class WebSocketClient implements AutoCloseable {
|
||||
webSocketClient.getProperties().put(ClientProperties.SSL_ENGINE_CONFIGURATOR, sslEngine);
|
||||
webSocketClient.getProperties().put(ClientProperties.INCOMING_BUFFER_SIZE, bufferSize);
|
||||
String wssPath = "wss://" + timelyHostname + ":" + timelyWssPort + "/websocket";
|
||||
session = webSocketClient.connectToServer(clientEndpoint, new TimelyEndpointConfig(clientEndpoint, sessionCookie), new URI(wssPath));
|
||||
session = webSocketClient.connectToServer(clientEndpoint, new TimelyEndpointConfig(sessionCookie), new URI(
|
||||
wssPath));
|
||||
|
||||
final ByteBuffer pingData = ByteBuffer.allocate(0);
|
||||
webSocketClient.getScheduledExecutorService().scheduleAtFixedRate(() -> {
|
||||
try {
|
||||
session.getBasicRemote().sendPing(pingData);
|
||||
} catch (Exception e) {
|
||||
log.error("Error sending ping", e);
|
||||
LOG.error("Error sending ping", e);
|
||||
}
|
||||
}, 30, 60, TimeUnit.SECONDS);
|
||||
closed = false;
|
||||
|
||||
+18
-18
@@ -18,30 +18,33 @@ import timely.api.AddSubscription;
|
||||
import timely.api.CloseSubscription;
|
||||
import timely.api.CreateSubscription;
|
||||
import timely.api.RemoveSubscription;
|
||||
import timely.api.VersionRequest;
|
||||
import timely.client.websocket.ClientHandler;
|
||||
import timely.client.websocket.WebSocketClient;
|
||||
import timely.serialize.JsonSerializer;
|
||||
|
||||
public class WebSocketSubscriptionClient extends WebSocketClient {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(WebSocketSubscriptionClient.class);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(WebSocketSubscriptionClient.class);
|
||||
private final String subscriptionId;
|
||||
|
||||
public WebSocketSubscriptionClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin,
|
||||
boolean hostVerificationEnabled, int bufferSize) {
|
||||
super(ssl, timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, hostVerificationEnabled, bufferSize);
|
||||
public WebSocketSubscriptionClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort,
|
||||
boolean doLogin, String timelyUsername, String timelyPassword, boolean hostVerificationEnabled,
|
||||
int bufferSize) {
|
||||
super(ssl, timelyHostname, timelyHttpsPort, timelyWssPort, doLogin, timelyUsername, timelyPassword,
|
||||
hostVerificationEnabled, bufferSize);
|
||||
subscriptionId = UUID.randomUUID().toString();
|
||||
log.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
|
||||
LOG.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
|
||||
}
|
||||
|
||||
public WebSocketSubscriptionClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String keyStoreFile,
|
||||
String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
|
||||
boolean hostVerificationEnabled, int bufferSize) {
|
||||
super(timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, keyStoreFile, keyStoreType, keyStorePass, trustStoreFile, trustStoreType,
|
||||
trustStorePass, hostVerificationEnabled, bufferSize);
|
||||
public WebSocketSubscriptionClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean doLogin,
|
||||
String timelyUsername, String timelyPassword, String keyStoreFile, String keyStoreType,
|
||||
String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
|
||||
boolean hostVerificationEnabled, int bufferSize) {
|
||||
super(timelyHostname, timelyHttpsPort, timelyWssPort, doLogin, timelyUsername, timelyPassword, keyStoreFile,
|
||||
keyStoreType, keyStorePass, trustStoreFile, trustStoreType, trustStorePass, hostVerificationEnabled,
|
||||
bufferSize);
|
||||
subscriptionId = UUID.randomUUID().toString();
|
||||
log.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
|
||||
LOG.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -61,13 +64,14 @@ public class WebSocketSubscriptionClient extends WebSocketClient {
|
||||
session.getBasicRemote().sendText(JsonSerializer.getObjectMapper().writeValueAsString(close));
|
||||
session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "Client called close."));
|
||||
} catch (Exception e) {
|
||||
log.info("Unable to send close message to server: {}", e.getMessage());
|
||||
LOG.info("Unable to send close message to server: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
super.close();
|
||||
}
|
||||
|
||||
public Future<Void> addSubscription(String metric, Map<String,String> tags, long startTime, long endTime, long delayTime) throws IOException {
|
||||
public Future<Void> addSubscription(String metric, Map<String, String> tags, long startTime, long endTime,
|
||||
long delayTime) throws IOException {
|
||||
AddSubscription add = new AddSubscription();
|
||||
add.setSubscriptionId(subscriptionId);
|
||||
add.setMetric(metric);
|
||||
@@ -85,8 +89,4 @@ public class WebSocketSubscriptionClient extends WebSocketClient {
|
||||
return session.getAsyncRemote().sendText(JsonSerializer.getObjectMapper().writeValueAsString(remove));
|
||||
}
|
||||
|
||||
public Future<Void> requestVersion() throws Exception {
|
||||
VersionRequest version = new VersionRequest();
|
||||
return session.getAsyncRemote().sendText(JsonSerializer.getObjectMapper().writeValueAsString(version));
|
||||
}
|
||||
}
|
||||
+11
-8
@@ -2,8 +2,10 @@ package timely.model;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.TreeSet;
|
||||
|
||||
import timely.model.parse.TagParser;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonGetter;
|
||||
import com.fasterxml.jackson.annotation.JsonInclude;
|
||||
@@ -11,17 +13,16 @@ import com.fasterxml.jackson.annotation.JsonPropertyOrder;
|
||||
import com.fasterxml.jackson.annotation.JsonRootName;
|
||||
import com.fasterxml.jackson.annotation.JsonUnwrapped;
|
||||
|
||||
import timely.model.parse.TagParser;
|
||||
|
||||
/**
|
||||
* A Metric consists of a metric name, tags, and a Value
|
||||
*
|
||||
* Json has been flattened to reduce overhead. Example: {"name":"m1","timestamp":1,"measure":1.0,"tags":[{"k1":"v1"}]}
|
||||
* Json has been flattened to reduce overhead. Example:
|
||||
* {"name":"m1","timestamp":1,"measure":1.0,"tags":[{"k1":"v1"}]}
|
||||
*
|
||||
*/
|
||||
@JsonRootName("metric")
|
||||
@JsonInclude(JsonInclude.Include.NON_NULL)
|
||||
@JsonPropertyOrder({"name"})
|
||||
@JsonPropertyOrder({ "name" })
|
||||
public class Metric implements Serializable {
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
@@ -79,7 +80,7 @@ public class Metric implements Serializable {
|
||||
}
|
||||
|
||||
public void setTags(List<Tag> tags) {
|
||||
this.tags = tags;
|
||||
this.tags.addAll(tags);
|
||||
}
|
||||
|
||||
public void addTag(Tag tag) {
|
||||
@@ -97,7 +98,8 @@ public class Metric implements Serializable {
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "Metric{" + "name='" + name + '\'' + ", tags=" + new TreeSet<>(tags) + ", value=" + value + '}';
|
||||
Collections.sort(tags);
|
||||
return "Metric{" + "name='" + name + '\'' + ", tags=" + tags + ", value=" + value + '}';
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -133,7 +135,8 @@ public class Metric implements Serializable {
|
||||
}
|
||||
|
||||
/**
|
||||
* Simple Builder for Metric. Note: cannot re-use a builder after build is called.
|
||||
* Simple Builder for Metric. Note: cannot re-use a builder after build is
|
||||
* called.
|
||||
*/
|
||||
public static class Builder {
|
||||
|
||||
+28
-14
@@ -2,13 +2,12 @@ package timely.model;
|
||||
|
||||
import java.lang.reflect.Array;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.Modifier;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.Stack;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
/**
|
||||
* A simple interface that objects can implement to return the object size.
|
||||
@@ -48,7 +47,7 @@ public interface ObjectSizeOf {
|
||||
|
||||
public static class Sizer {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(Sizer.class);
|
||||
private static final Logger log = Logger.getLogger(Sizer.class.getCanonicalName());
|
||||
public static final short OBJECT_OVERHEAD = 8;
|
||||
public static final short ARRAY_OVERHEAD = 12;
|
||||
public static final short REFERENCE = 4;
|
||||
@@ -57,14 +56,19 @@ public interface ObjectSizeOf {
|
||||
public static final short NUMBER_SIZE = 16;
|
||||
|
||||
/**
|
||||
* Get the size of an object. Note that we want something relatively fast that gives us an order of magnitude here. The java Instrumentation agent
|
||||
* mechanism is a little too costly for general use here. This will look for the ObjectSizeOf interface and if implemented on the object will use that.
|
||||
* Otherwise it will do a simple navigation of the fields using reflection.
|
||||
* Get the size of an object. Note that we want something relatively
|
||||
* fast that gives us an order of magnitude here. The java
|
||||
* Instrumentation agent mechanism is a little too costly for general
|
||||
* use here. This will look for the ObjectSizeOf interface and if
|
||||
* implemented on the object will use that. Otherwise it will do a
|
||||
* simple navigation of the fields using reflection.
|
||||
*
|
||||
* @param o
|
||||
* @return an approximation of the object size
|
||||
*/
|
||||
public static long getObjectSize(Object o, boolean useSizeInBytesMethod, boolean debug) {
|
||||
long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(), useSizeInBytesMethod, debug);
|
||||
long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(),
|
||||
useSizeInBytesMethod, debug);
|
||||
return estSize;
|
||||
}
|
||||
|
||||
@@ -73,7 +77,8 @@ public interface ObjectSizeOf {
|
||||
return estSize;
|
||||
}
|
||||
|
||||
public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack, boolean useSizeInBytesMethod, boolean debug) {
|
||||
public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack,
|
||||
boolean useSizeInBytesMethod, boolean debug) {
|
||||
long totalSize = 0;
|
||||
stack.add(new ObjectInstance(o));
|
||||
while (!stack.isEmpty()) {
|
||||
@@ -84,16 +89,25 @@ public interface ObjectSizeOf {
|
||||
visited.add(oi);
|
||||
try {
|
||||
if (useSizeInBytesMethod) {
|
||||
if (o instanceof ObjectSizeOf) {
|
||||
size = ((ObjectSizeOf) o).sizeInBytes();
|
||||
try {
|
||||
if (o instanceof ObjectSizeOf) {
|
||||
size = ((ObjectSizeOf) o).sizeInBytes();
|
||||
} else {
|
||||
Method sizeInBytes = o.getClass().getMethod("sizeInBytes", (Class<?>[]) null);
|
||||
size = (Long) sizeInBytes.invoke(o);
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
// ok, lets do this the hard way...
|
||||
}
|
||||
}
|
||||
if (size == 0) {
|
||||
// the hard way...
|
||||
// do not include Class related objects or
|
||||
// reflection objects
|
||||
if ((o instanceof Class) || (o instanceof ClassLoader)
|
||||
|| (o.getClass().getPackage() != null && o.getClass().getPackage().getName().startsWith("java.lang.reflect"))) {
|
||||
if ((o instanceof Class)
|
||||
|| (o instanceof ClassLoader)
|
||||
|| (o.getClass().getPackage() != null && o.getClass().getPackage().getName()
|
||||
.startsWith("java.lang.reflect"))) {
|
||||
size = 0;
|
||||
} else if (o instanceof Number || o instanceof Boolean || o instanceof Character) {
|
||||
size = NUMBER_SIZE;
|
||||
@@ -148,7 +162,7 @@ public interface ObjectSizeOf {
|
||||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
log.warn("Unable to determine object size for " + o, t);
|
||||
log.warning("Unable to determine object size for " + o);
|
||||
}
|
||||
if (debug) {
|
||||
System.out.println(o.getClass().getCanonicalName() + "=" + size);
|
||||
+12
-11
@@ -1,21 +1,19 @@
|
||||
package timely.model;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.apache.commons.lang3.builder.CompareToBuilder;
|
||||
import org.apache.commons.lang3.builder.EqualsBuilder;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonAnyGetter;
|
||||
import com.fasterxml.jackson.annotation.JsonAnySetter;
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import com.fasterxml.jackson.annotation.JsonRootName;
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
import org.apache.commons.lang3.builder.CompareToBuilder;
|
||||
import org.apache.commons.lang3.builder.EqualsBuilder;
|
||||
import timely.model.parse.TagParser;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
/**
|
||||
* Tag consists of key value pair
|
||||
*/
|
||||
@@ -29,7 +27,8 @@ public class Tag implements Comparable<Tag>, Serializable {
|
||||
private String key;
|
||||
private String value;
|
||||
|
||||
public Tag() {}
|
||||
public Tag() {
|
||||
}
|
||||
|
||||
public Tag(Tag other) {
|
||||
this.setKey(other.getKey());
|
||||
@@ -48,7 +47,7 @@ public class Tag implements Comparable<Tag>, Serializable {
|
||||
}
|
||||
|
||||
@JsonAnyGetter
|
||||
public Map<String,String> get() {
|
||||
public Map<String, String> get() {
|
||||
return Stream.of(this).collect(Collectors.toMap(Tag::getKey, Tag::getValue));
|
||||
}
|
||||
|
||||
@@ -78,6 +77,8 @@ public class Tag implements Comparable<Tag>, Serializable {
|
||||
|
||||
/**
|
||||
* Joins key and value into key=value
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public String join() {
|
||||
return tagParser.combine(this);
|
||||
+10
-7
@@ -1,13 +1,13 @@
|
||||
package timely.model;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Comparator;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonCreator;
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import com.google.common.base.Objects;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Comparator;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* Value consists of a long timestamp and a double measurement
|
||||
*/
|
||||
@@ -18,7 +18,8 @@ public class Value implements Comparator<Value>, Serializable {
|
||||
private Long timestamp;
|
||||
private Double measure;
|
||||
|
||||
public Value() {}
|
||||
public Value() {
|
||||
}
|
||||
|
||||
public Value(Value other) {
|
||||
this.setTimestamp(other.getTimestamp());
|
||||
@@ -73,9 +74,11 @@ public class Value implements Comparator<Value>, Serializable {
|
||||
if (null == o || getClass() != o.getClass())
|
||||
return false;
|
||||
|
||||
if (getTimestamp() != null ? !getTimestamp().equals(((Value) o).getTimestamp()) : ((Value) o).getTimestamp() != null)
|
||||
if (getTimestamp() != null ? !getTimestamp().equals(((Value) o).getTimestamp())
|
||||
: ((Value) o).getTimestamp() != null)
|
||||
return false;
|
||||
return !(getMeasure() != null ? !getMeasure().equals(((Value) o).getMeasure()) : ((Value) o).getMeasure() != null);
|
||||
return !(getMeasure() != null ? !getMeasure().equals(((Value) o).getMeasure())
|
||||
: ((Value) o).getMeasure() != null);
|
||||
}
|
||||
|
||||
@Override
|
||||
File renamed without changes.
+2
-3
@@ -1,12 +1,11 @@
|
||||
package timely.model.parse;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.base.Splitter;
|
||||
|
||||
import timely.model.Metric;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class MetricParser implements Parser<Metric> {
|
||||
|
||||
private static final Splitter spaceSplitter = Splitter.on(" ").omitEmptyStrings().trimResults();
|
||||
File renamed without changes.
+12
-17
@@ -1,34 +1,27 @@
|
||||
package timely.model.parse;
|
||||
|
||||
import com.google.common.base.Splitter;
|
||||
import timely.model.Tag;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import timely.model.Tag;
|
||||
|
||||
/**
|
||||
* Parses tag1=val1,tag2=val2,tag3=val3 into a {@code List} of {@code Tag} Combines {@code List} of {@code Tag} into {@code String} of
|
||||
* Parses tag1=val1,tag2=val2,tag3=val3 into a {@code List} of {@code Tag}
|
||||
* Combines {@code List} of {@code Tag} into {@code String} of
|
||||
* tag1=val1,tag2=val2,tag3=val3.
|
||||
*/
|
||||
public class TagListParser implements Parser<List<Tag>>, Combiner<Collection<Tag>> {
|
||||
|
||||
private static final Splitter commaSplitter = Splitter.on(",").omitEmptyStrings();
|
||||
private static final TagParser tagParser = new TagParser();
|
||||
|
||||
@Override
|
||||
public List<Tag> parse(String t) {
|
||||
ArrayList<Tag> tags = new ArrayList<>();
|
||||
// only split on commas that are not preceded with a backslash
|
||||
String[] parts = t.split("(?<!\\\\),");
|
||||
for (String p : parts) {
|
||||
if (!p.isEmpty()) {
|
||||
Tag tag = tagParser.parse(p);
|
||||
// unescape commas in the key and value of the tag
|
||||
tag.setKey(tag.getKey().replaceAll("\\\\,", ","));
|
||||
tag.setValue(tag.getValue().replaceAll("\\\\,", ","));
|
||||
tags.add(tag);
|
||||
}
|
||||
}
|
||||
commaSplitter.splitToList(t).forEach(p -> tags.add(tagParser.parse(p)));
|
||||
return tags;
|
||||
}
|
||||
|
||||
@@ -43,11 +36,13 @@ public class TagListParser implements Parser<List<Tag>>, Combiner<Collection<Tag
|
||||
}
|
||||
|
||||
/**
|
||||
* An alternative combiner for a map of string to string instead of a list of tags
|
||||
*
|
||||
* An alternative combiner for a map of string to string instead of a list
|
||||
* of tags
|
||||
*
|
||||
* @param tags
|
||||
* @return string representation
|
||||
*/
|
||||
public String combine(Map<String,String> tags) {
|
||||
public String combine(Map<String, String> tags) {
|
||||
StringBuilder builder = new StringBuilder();
|
||||
if (!tags.isEmpty()) {
|
||||
tags.forEach((k, v) -> builder.append(tagParser.combine(new Tag(k, v))).append(','));
|
||||
+5
-5
@@ -1,18 +1,18 @@
|
||||
package timely.model.parse;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import com.google.common.base.Joiner;
|
||||
import com.google.common.base.Splitter;
|
||||
|
||||
import timely.model.Tag;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Parses tag1=val1 into a {@code Tag} Combines {@code Tag} into {@code String} of tag1=val1
|
||||
* Parses tag1=val1 into a {@code Tag} Combines {@code Tag} into {@code String}
|
||||
* of tag1=val1
|
||||
*/
|
||||
public class TagParser implements Parser<Tag>, Combiner<Tag> {
|
||||
|
||||
private static final Splitter equalSplitter = Splitter.on("=").limit(2).trimResults();
|
||||
private static final Splitter equalSplitter = Splitter.on("=").trimResults();
|
||||
private static final Joiner equalsJoiner = Joiner.on("=");
|
||||
|
||||
@Override
|
||||
@@ -23,16 +23,12 @@ def pivot(df, metric, groupByColumn=None):
|
||||
return dataFrame
|
||||
|
||||
|
||||
def resample(df, sample, how='mean', interpolate=True, fill=None):
|
||||
def resample(df, sample, how='mean'):
|
||||
|
||||
dataFrame = pandas.DataFrame(df, copy=True)
|
||||
if dataFrame is not None:
|
||||
if sample is not None:
|
||||
dataFrame = dataFrame.resample(sample, how=how)
|
||||
if (interpolate == True):
|
||||
dataFrame = dataFrame.interpolate()
|
||||
if fill is not None:
|
||||
dataFrame = dataFrame.fillna(value=fill)
|
||||
dataFrame = dataFrame.resample(sample, how=how).interpolate()
|
||||
return dataFrame
|
||||
|
||||
|
||||
@@ -329,4 +325,4 @@ def getTitle(timelyMetric, config, separator='\n'):
|
||||
title += separator + alertTitle
|
||||
|
||||
|
||||
return title
|
||||
return title
|
||||
@@ -2,6 +2,7 @@ import time
|
||||
from datetime import datetime
|
||||
from datetime import timedelta
|
||||
from datetime import tzinfo
|
||||
from pandas.tslib import Timedelta
|
||||
import pandas
|
||||
|
||||
class TimeDateError(ValueError):
|
||||
@@ -40,7 +41,7 @@ class TimeDateRange:
|
||||
def getBeginMs(self):
|
||||
if self.beginDt == None:
|
||||
return None
|
||||
return TimeDateRange.unix_time_millis(self.beginDt)
|
||||
return int(time.mktime(self.beginDt.timetuple()) * 1000)
|
||||
|
||||
def getBeginStr(self):
|
||||
if self.beginDt == None:
|
||||
@@ -50,7 +51,7 @@ class TimeDateRange:
|
||||
def getEndMs(self):
|
||||
if self.endDt == None:
|
||||
return None
|
||||
return TimeDateRange.unix_time_millis(self.endDt)
|
||||
return int(time.mktime(self.endDt.timetuple()) * 1000)
|
||||
|
||||
def getEndStr(self):
|
||||
if self.endDt == None:
|
||||
@@ -122,6 +123,6 @@ class TimeDateRange:
|
||||
|
||||
@classmethod
|
||||
def unix_time_millis(cls, dt):
|
||||
return int((dt - TimeDateRange.epoch).total_seconds() * 1000.0)
|
||||
return (dt - TimeDateRange.epoch).total_seconds() * 1000.0
|
||||
|
||||
|
||||
@@ -87,7 +87,7 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
|
||||
graphDF = graphDF.select(lambda x : re.search(analyticConfig.includeColRegex, x), axis=1)
|
||||
|
||||
if analyticConfig.sample is not None:
|
||||
graphDF = TimelyMetric.resample(graphDF, analyticConfig.sample, how=analyticConfig.how, interpolate=analyticConfig.interpolate, fill=analyticConfig.fill)
|
||||
graphDF = TimelyMetric.resample(graphDF, analyticConfig.sample, how=analyticConfig.how)
|
||||
|
||||
graphDF_avg = pandas.DataFrame(graphDF, copy=True)
|
||||
|
||||
@@ -221,4 +221,4 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
|
||||
|
||||
timelyAlert = TimelyAlert(timelyMetric, combined, message, seriesConfig, alertAnalyticConfig, notebook)
|
||||
|
||||
return timelyAlert
|
||||
return timelyAlert
|
||||
@@ -13,8 +13,6 @@ class TimelyAnalyticConfiguration():
|
||||
self.counter = analyticConfig.get('counter', False)
|
||||
self.sample_period = analyticConfig.get('sample', None)
|
||||
self.how = analyticConfig.get('how', 'mean')
|
||||
self.interpolate = analyticConfig.get('interpolate', True)
|
||||
self.fill = analyticConfig.get('fill', None)
|
||||
self.rolling_average_period = analyticConfig.get('rolling_average_period', None)
|
||||
self.min_threshold = analyticConfig.get('min_threshold', None)
|
||||
self.average_min_threshold = analyticConfig.get('average_min_threshold', None)
|
||||
@@ -65,8 +63,6 @@ class TimelyAnalyticConfiguration():
|
||||
self.sample_minutes = analyticConfig.sample_minutes
|
||||
self.sample = analyticConfig.sample
|
||||
self.how = analyticConfig.how
|
||||
self.interpolate = analyticConfig.interpolate
|
||||
self.fill = analyticConfig.fill
|
||||
self.rolling_average_period = analyticConfig.rolling_average_period
|
||||
self.rolling_average_samples = analyticConfig.rolling_average_samples
|
||||
self.rolling_average_minutes = analyticConfig.rolling_average_minutes
|
||||
|
||||
@@ -16,15 +16,14 @@ utc = UTC()
|
||||
class TimelyWebSocketClient(WebSocketClient):
|
||||
|
||||
def __init__(self, hostport, metric, tags, startTime, endTime, connect_timeout=WebSocketClient.DEFAULT_CONNECT_TIMEOUT,
|
||||
request_timeout=WebSocketClient.DEFAULT_REQUEST_TIMEOUT, client_cert=None, client_key=None, ca_certs=None, validate_cert=False):
|
||||
request_timeout=WebSocketClient.DEFAULT_REQUEST_TIMEOUT):
|
||||
|
||||
self.metric = metric
|
||||
self.tags = tags
|
||||
self.startTime = startTime
|
||||
self.endTime = endTime
|
||||
self.subscriptionId = random.randint(1000000000, 9999999999)
|
||||
WebSocketClient.__init__(self, hostport, '/websocket', {'metric' : metric}, connect_timeout, request_timeout,
|
||||
client_cert=client_cert, client_key=client_key, ca_certs=ca_certs, validate_cert=validate_cert)
|
||||
WebSocketClient.__init__(self, hostport, connect_timeout, request_timeout)
|
||||
|
||||
def _on_message(self, msg):
|
||||
# implement in subclass
|
||||
@@ -38,6 +37,7 @@ class TimelyWebSocketClient(WebSocketClient):
|
||||
"subscriptionId": self.subscriptionId,
|
||||
}
|
||||
self.send(create)
|
||||
self.send(create)
|
||||
|
||||
m1 = {
|
||||
"operation": "add",
|
||||
@@ -73,8 +73,7 @@ class TimelyMetric(TimelyWebSocketClient):
|
||||
client = None
|
||||
endtime = None
|
||||
|
||||
def __init__(self, hostport, metric, tags, beginStr, endStr, period, sample,
|
||||
client_cert=None, client_key=None, ca_certs=None, validate_cert=False):
|
||||
def __init__(self, hostport, metric, tags, beginStr, endStr, period, sample):
|
||||
self.series = pandas.Series()
|
||||
self.metric = metric
|
||||
self.tags = tags
|
||||
@@ -84,8 +83,7 @@ class TimelyMetric(TimelyWebSocketClient):
|
||||
self.data = []
|
||||
self.timeDateRange = TimeDateRange(beginStr, endStr, period)
|
||||
|
||||
TimelyWebSocketClient.__init__(self, hostport, metric, tags, self.timeDateRange.getBeginMs(), self.timeDateRange.getEndMs(),
|
||||
client_cert=client_cert, client_key=client_key, ca_certs=ca_certs, validate_cert=validate_cert)
|
||||
TimelyWebSocketClient.__init__(self, hostport, metric, tags, self.timeDateRange.getBeginMs(), self.timeDateRange.getEndMs())
|
||||
|
||||
def debugOn(self):
|
||||
self.debug = True
|
||||
@@ -107,41 +105,30 @@ class TimelyMetric(TimelyWebSocketClient):
|
||||
global endtime
|
||||
|
||||
responsesObj = json.loads(msg)
|
||||
|
||||
responses = responsesObj.get("responses")
|
||||
if responses is not None:
|
||||
for obj in responses:
|
||||
complete = bool(obj.get("complete"))
|
||||
if complete:
|
||||
self._on_connection_close()
|
||||
return
|
||||
else:
|
||||
date = int(obj.get("timestamp")/1000)
|
||||
dt = pandas.datetime.utcfromtimestamp(date)
|
||||
metricName = str(obj.get("metric"))
|
||||
metricValue = obj.get("value")
|
||||
|
||||
newData = {}
|
||||
newData["date"] = dt
|
||||
newData[metricName] = metricValue
|
||||
for obj in responses:
|
||||
complete = bool(obj.get("complete"))
|
||||
if complete:
|
||||
self._on_connection_close()
|
||||
return
|
||||
else:
|
||||
date = int(obj.get("timestamp")/1000)
|
||||
dt = pandas.datetime.utcfromtimestamp(date)
|
||||
metricName = str(obj.get("metric"))
|
||||
metricValue = obj.get("value")
|
||||
|
||||
tags = obj.get("tags")
|
||||
for d in tags:
|
||||
for k,v in d.items():
|
||||
newData[k] = v
|
||||
newData = {}
|
||||
newData["date"] = dt
|
||||
newData[metricName] = metricValue
|
||||
|
||||
tags = obj.get("tags")
|
||||
for d in tags:
|
||||
for k,v in d.items():
|
||||
newData[k] = v
|
||||
|
||||
self.data.append(newData)
|
||||
|
||||
self.data.append(newData)
|
||||
else:
|
||||
print("Session ended abnormally")
|
||||
exception = responsesObj.get("TimelyException")
|
||||
if exception is not None:
|
||||
code = exception.get("code")
|
||||
message = exception.get("message")
|
||||
print("Exception")
|
||||
print("code: " + str(code))
|
||||
print("message: " + message)
|
||||
self._on_connection_close()
|
||||
return
|
||||
|
||||
def _on_connection_close(self):
|
||||
global client
|
||||
@@ -150,8 +137,6 @@ class TimelyMetric(TimelyWebSocketClient):
|
||||
if len(self.dataFrame) > 0:
|
||||
self.dataFrame['date'] = pandas.DatetimeIndex(self.dataFrame['date'])
|
||||
self.dataFrame = self.dataFrame.set_index('date')
|
||||
j = pandas.to_timedelta(self.dataFrame.groupby(self.dataFrame.index).cumcount(), unit='us').values
|
||||
self.dataFrame.index = self.dataFrame.index + j
|
||||
|
||||
self.close()
|
||||
ioloop.IOLoop.current(instance=False).stop()
|
||||
|
||||
@@ -14,37 +14,18 @@ class WebSocketClient():
|
||||
DEFAULT_REQUEST_TIMEOUT = 60
|
||||
APPLICATION_JSON = 'application/json'
|
||||
|
||||
def __init__(self, hostport, path, params, connect_timeout=DEFAULT_CONNECT_TIMEOUT,
|
||||
request_timeout=DEFAULT_REQUEST_TIMEOUT, client_cert=None, client_key=None, ca_certs=None, validate_cert=False):
|
||||
def __init__(self, hostport, connect_timeout=DEFAULT_CONNECT_TIMEOUT,
|
||||
request_timeout=DEFAULT_REQUEST_TIMEOUT):
|
||||
|
||||
self.connect_timeout = connect_timeout
|
||||
self.request_timeout = request_timeout
|
||||
self.client_cert=client_cert
|
||||
self.client_key=client_key
|
||||
self.ca_certs=ca_certs
|
||||
self.validate_cert=validate_cert
|
||||
self.url = 'wss://' + hostport
|
||||
if (path is not None and len(path) > 0):
|
||||
if (path.startswith('/')):
|
||||
self.url = self.url + path
|
||||
else:
|
||||
self.url = self.url + '/' + path
|
||||
if (any(params.values())):
|
||||
first = True
|
||||
for key, value in params.iteritems():
|
||||
if (first == True):
|
||||
self.url = self.url + '?'
|
||||
first = False
|
||||
else:
|
||||
self.url = self.url + '&'
|
||||
self.url = self.url + key + '=' + value
|
||||
self.url = 'wss://' + hostport + '/websocket'
|
||||
|
||||
def connect(self, ioloop=ioloop.IOLoop.current(instance=False)):
|
||||
"""Connect to the server.
|
||||
:param str url: server URL.
|
||||
"""
|
||||
request = websocket.httpclient.HTTPRequest(self.url, connect_timeout=self.connect_timeout, request_timeout=self.request_timeout, validate_cert=self.validate_cert,
|
||||
client_cert=self.client_cert, client_key=self.client_key, ca_certs=self.ca_certs)
|
||||
request = websocket.httpclient.HTTPRequest(self.url, connect_timeout=self.connect_timeout, request_timeout=self.request_timeout, validate_cert=False)
|
||||
websocket.websocket_connect(request, io_loop=ioloop, callback=self._connect_callback)
|
||||
|
||||
def send(self, data):
|
||||
|
||||
Loaded 100 of 629 files, more files were not shown because too many files have changed in this diff.
Show more
Reference in new issue
Block a user