Compare commits

..
20 Commits
Author SHA1 Message Date
Bill Oley d35ae08188 Tag version 1.2.4 2019-05-22 17:14:42 -04:00
Bill Oley e448657afb Correction in websocket pipeline from Netty API change 2019-05-22 16:49:11 -04:00
Bill Oley 1dcd893711 Add age off iterator for meta table
Add age off iterator for meta table
2019-05-22 16:49:08 -04:00
Bill Oley 409f5375d1 Change pom version to 1.2.4-SNAPSHOT 2019-05-22 16:48:09 -04:00
Bill Oley c5df8139ca Correct typos 2019-05-22 16:46:55 -04:00
Bill Oley dcd23d4fc8 fixes #181 Use netty-tcnative-boringssl-static 2019-04-18 11:56:54 -04:00
Bill Oley cc7326fc19 Tag version 1.2.2 2019-04-12 11:24:53 -04:00
Bill Oley 0d2b6e61c8 Correct typos 2019-04-12 10:55:04 -04:00
Bill Oley b7cd8765c5 Correct typos and minor changed to MetaCache 2019-04-10 15:31:53 -04:00
Bill Oley a9d27d5b1b re #179 Allow AuthCache to periodically refresh credentials, cache meta table up to a configurable number of tags 2019-04-08 10:50:14 -04:00
Bill Oley 7d67eb218d Update netty-tcnative dependency, remove x509 principal extractor from AuthenticationService 2019-04-03 15:52:19 -04:00
Bill Oley 5fb7e3dd2b Bump pom versions to 1.2.2-SNAPSHOT 2019-04-02 12:57:22 -04:00
Bill Oley 5a89861cec Tag version 1.2.1 2019-04-02 12:53:04 -04:00
Bill Oley a3bc9a01ee Update dependency versions
Update dependencies

Update dependency versions
2019-04-02 12:43:36 -04:00
Bill Oley dba494b06d Bump pom versions to 1.2.1-SNAPSHOT 2019-02-25 15:41:31 -05:00
Bill Oley 5826884c3c Tag version 1.2.0 2019-02-25 15:11:57 -05:00
Bill Oley f94c731bbc Remove unused modeule directory introduced in error 2019-02-25 15:11:12 -05:00
Bill Oley 4e042bcb94 Allow all http methods in grafana auth proxy, disable cookies in HttpPools 2019-02-22 16:17:36 -05:00
Bill Oley ab7f499134 closes #180 Allow a configurable default visibility for metrics 2019-02-21 15:33:25 -05:00
Bill Oley c098f9f3e4 re #179 Create an authentication app for grafana that uses Timely's authentication framework and grafana's auth-proxy
re #179 Create authentication proxy for grafana
2019-02-21 13:28:12 -05:00
75 changed files with 1129 additions and 1137 deletions

No files matched your search

+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>timely-analytics</artifactId>
<!--
+3 -4
View File
@@ -27,15 +27,14 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
$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="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
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} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
#JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
+5 -2
View File
@@ -46,9 +46,12 @@
<user-service id="usernamePasswordService">
<user name="test" password="test1" authorities="A,B,C" />
</user-service>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:bean id="noOpPasswordEncoder" class="org.springframework.security.crypto.password.NoOpPasswordEncoder"/>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:property name="userDetailsService" ref="usernamePasswordService"/>
<beans:property name="passwordEncoder" ref="noOpPasswordEncoder"/>
</beans:bean>
<authentication-manager alias="authenticationManager">
+2
View File
@@ -8,6 +8,8 @@ timely-balancer:
assignmentFile: /timely/balancer/assignments.csv
security:
session-max-age: 86400
cache-expiration-minutes: 60
cache-refresh-minutes: 5
allow-anonymous-http-access: true
allow-anonymous-ws-access: true
serverSsl:
+3 -38
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>timely-balancer</artifactId>
<dependencies>
@@ -161,9 +161,8 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>runtime</scope>
</dependency>
<dependency>
@@ -247,9 +246,8 @@
<artifactItems>
<artifactItem>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<type>jar</type>
<overWrite>false</overWrite>
<outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -291,37 +289,4 @@
</extension>
</extensions>
</build>
<profiles>
<profile>
<id>native-epoll-linux</id>
<activation>
<os>
<family>!mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
<classifier>linux-x86_64</classifier>
</dependency>
</dependencies>
</profile>
<profile>
<id>native-epoll</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
@@ -64,6 +64,7 @@ import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import timely.auth.AuthCache;
@@ -143,7 +144,7 @@ public class Balancer {
}
protected static ConfigurableApplicationContext initializeConfiguration(String[] args) {
return new SpringApplicationBuilder(SpringBootstrap.class).web(false).run(args);
return new SpringApplicationBuilder(SpringBootstrap.class).web(WebApplicationType.NONE).run(args);
}
private void shutdownHook() {
@@ -363,7 +364,7 @@ public class Balancer {
curatorFramework.start();
ensureZkPaths(curatorFramework, zkPaths);
AuthCache.setSessionMaxAge(balancerConfig.getSecurity());
AuthCache.configure(balancerConfig.getSecurity());
final boolean useEpoll = useEpoll();
Class<? extends ServerSocketChannel> channelClass;
Class<? extends Channel> datagramChannelClass;
@@ -510,7 +511,7 @@ public class Balancer {
ch.pipeline().addLast("decoder", new HttpRequestDecoder());
ch.pipeline().addLast("compressor", new HttpContentCompressor());
ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
ch.pipeline().addLast("queryDecoder", new timely.netty.http.HttpRequestDecoder(
balancerConfig.getSecurity(), balancerConfig.getHttp()));
@@ -531,7 +532,7 @@ public class Balancer {
ch.pipeline().addLast("logger", new LoggingHandler());
ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
ch.pipeline().addLast("putDecoder", new UdpDecoder());
ch.pipeline().addLast("udpRelayHandler", new UdpRelayHandler(metricResolver, udpClientPool));
}
@@ -544,7 +545,7 @@ public class Balancer {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
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());
@@ -560,13 +561,13 @@ public class Balancer {
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(8192));
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
ch.pipeline().addLast("sessionExtractor", new WebSocketFullRequestHandler());
ch.pipeline().addLast("idle-handler",
new IdleStateHandler(balancerConfig.getWebsocket().getTimeout(), 0, 0));
ch.pipeline().addLast("ws-protocol",
new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, true));
new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, false, true));
ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(balancerConfig.getSecurity()));
ch.pipeline().addLast("httpRelay", new WsRelayHandler(balancerConfig, metricResolver, wsClientPool));
ch.pipeline().addLast("error", new WSTimelyExceptionHandler());
@@ -5,8 +5,10 @@ import javax.validation.Valid;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.stereotype.Component;
import org.springframework.validation.annotation.Validated;
import timely.configuration.Cache;
@Validated
@Component
@ConfigurationProperties(prefix = "timely-balancer")
public class BalancerConfiguration {
@@ -5,7 +5,6 @@ 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.BasicCookieStore;
import org.apache.http.impl.client.CloseableHttpClient;
import timely.balancer.configuration.BalancerSecurity;
import timely.balancer.connection.TimelyBalancedHost;
@@ -22,8 +21,8 @@ public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalance
@Override
public PooledObject<CloseableHttpClient> makeObject(TimelyBalancedHost k) throws Exception {
BasicCookieStore cookieJar = new BasicCookieStore();
return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, cookieJar,
// disable cookie management because we are sharing connections
return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, null,
security.getClientSsl().isHostVerificationEnabled(), security.getClientSsl().isUseClientCert()));
}
+3 -2
View File
@@ -4,12 +4,13 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>timely-client</artifactId>
<properties>
<!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
<python.client.version>1.1.1.dev0</python.client.version>
<!-- use .dev0 instead of -SNAPSHOT -->
<python.client.version>1.2.4</python.client.version>
</properties>
<dependencies>
<dependency>
@@ -46,15 +46,21 @@ public class HttpClient {
public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled,
boolean clientAuth) {
RequestConfig defaultRequestConfig = RequestConfig.custom().setCookieSpec(CookieSpecs.STANDARD).build();
HostnameVerifier hostnameVerifier;
if (hostVerificationEnabled) {
hostnameVerifier = SSLConnectionSocketFactory.getDefaultHostnameVerifier();
} else {
hostnameVerifier = new NoopHostnameVerifier();
}
HttpClientBuilder builder = HttpClients.custom().setSSLContext(ssl).setDefaultCookieStore(cookieStore)
.setDefaultRequestConfig(defaultRequestConfig).setSSLHostnameVerifier(hostnameVerifier);
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,
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>collectd-nsq-plugin</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>collectd-plugin-parent</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>collectd-timely-plugin</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>timely-common</artifactId>
<dependencies>
+3 -4
View File
@@ -27,15 +27,14 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
$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="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
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} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
#JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
+2
View File
@@ -4,6 +4,8 @@ grafana-auth:
port: 3001
security:
session-max-age: 86400
cache-expiration-minutes: 60
cache-refresh-minutes: 5
allow-anonymous-http-access: true
allow-anonymous-ws-access: true
serverSsl:
+5 -2
View File
@@ -46,9 +46,12 @@
<user-service id="usernamePasswordService">
<user name="test" password="test1" authorities="A,B,C" />
</user-service>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:bean id="noOpPasswordEncoder" class="org.springframework.security.crypto.password.NoOpPasswordEncoder"/>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:property name="userDetailsService" ref="usernamePasswordService"/>
<beans:property name="passwordEncoder" ref="noOpPasswordEncoder"/>
</beans:bean>
<authentication-manager alias="authenticationManager">
+3 -38
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>grafana-auth</artifactId>
<dependencies>
@@ -165,9 +165,8 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>runtime</scope>
</dependency>
<dependency>
@@ -251,9 +250,8 @@
<artifactItems>
<artifactItem>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<type>jar</type>
<overWrite>false</overWrite>
<outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -295,37 +293,4 @@
</extension>
</extensions>
</build>
<profiles>
<profile>
<id>native-epoll-linux</id>
<activation>
<os>
<family>!mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
<classifier>linux-x86_64</classifier>
</dependency>
</dependencies>
</profile>
<profile>
<id>native-epoll</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
+1
View File
@@ -20,6 +20,7 @@
<outputDirectory>lib</outputDirectory>
<includes>
<include>timely*.jar</include>
<include>grafana*.jar</include>
</includes>
</fileSet>
<fileSet>
@@ -43,6 +43,7 @@ import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.internal.SystemPropertyUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import timely.auth.AuthCache;
@@ -81,7 +82,7 @@ public class GrafanaAuth {
}
protected static ConfigurableApplicationContext initializeConfiguration(String[] args) {
return new SpringApplicationBuilder(SpringBootstrap.class).web(false).run(args);
return new SpringApplicationBuilder(SpringBootstrap.class).web(WebApplicationType.NONE).run(args);
}
private void shutdownHook() {
@@ -225,7 +226,7 @@ public class GrafanaAuth {
public void run() throws Exception {
AuthCache.setSessionMaxAge(grafanaAuthConfig.getSecurity());
AuthCache.configure(grafanaAuthConfig.getSecurity());
final boolean useEpoll = useEpoll();
Class<? extends ServerSocketChannel> channelClass;
if (useEpoll) {
@@ -304,7 +305,9 @@ public class GrafanaAuth {
ch.pipeline().addLast("decoder", new HttpRequestDecoder());
ch.pipeline().addLast("compressor", new HttpContentCompressor());
ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
// high maximum contentLength so that grafana snapshots can be delivered
// might not be necessary if inbound chunking (while proxying) is handled
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(2097152));
ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
ch.pipeline().addLast("grafanaDecoder",
new GrafanaRequestDecoder(config.getSecurity(), config.getHttp()));
@@ -4,12 +4,29 @@ public class Grafana {
private String host = "localhost";
private Integer port = 3000;
private String protocol = "http";
public void setHost(String host) {
this.host = host;
}
public String getHost() {
return host;
}
public void setPort(Integer port) {
this.port = port;
}
public Integer getPort() {
return port;
}
public void setProtocol(String protocol) {
this.protocol = protocol;
}
public String getProtocol() {
return protocol;
}
}
@@ -5,9 +5,11 @@ import javax.validation.Valid;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.stereotype.Component;
import org.springframework.validation.annotation.Validated;
import timely.balancer.configuration.BalancerHttp;
import timely.balancer.configuration.BalancerSecurity;
@Validated
@Component
@ConfigurationProperties(prefix = "grafana-auth")
public class GrafanaAuthConfiguration {
@@ -22,7 +24,7 @@ public class GrafanaAuthConfiguration {
@Valid
@NestedConfigurationProperty
private Grafana grafana;
private Grafana grafana = new Grafana();
private Integer shutdownQuietPeriod = 5;
@@ -4,6 +4,7 @@ import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
import java.io.ByteArrayOutputStream;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -15,13 +16,20 @@ 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 org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.client.methods.HttpHead;
import org.apache.http.client.methods.HttpOptions;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.message.BasicHeader;
import org.slf4j.Logger;
@@ -59,6 +67,7 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
FullHttpRequest request = null;
GrafanaRequest grafanaRequest;
TimelyBalancedHost k = null;
String relayURI = null;
try {
try {
request = msg.getHttpRequest();
@@ -75,6 +84,7 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
throw new IllegalStateException("No token on grafana request");
}
String protocol = config.getGrafana().getProtocol();
String host = config.getGrafana().getHost();
Integer port = config.getGrafana().getPort();
k = TimelyBalancedHost.of(host, -1, port, -1, -1);
@@ -87,10 +97,8 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
TimelyPrincipal principal = token.getTimelyPrincipal();
TimelyUser user = principal.getPrimaryUser();
String subjectDn = user.getDn().subjectDN();
int equalsIndex = subjectDn.indexOf('=');
int endIndex = subjectDn.lastIndexOf(' ');
String userName = DnUtils.getCommonName(subjectDn);
String loginName = DnUtils.getShortName(subjectDn);
String userName = equalsIndex > -1 ? subjectDn.substring(equalsIndex + 1, endIndex) : subjectDn;
headers.put("X-WEBAUTH-USER", loginName);
headers.put("X-WEBAUTH-NAME", userName);
List<Header> relayedHeaderList = new ArrayList<>();
@@ -101,13 +109,50 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
}
Header[] headerArray = new Header[relayedHeaderList.size()];
relayedHeaderList.toArray(headerArray);
String relayURI = "https://" + k.getHost() + ":" + k.getHttpPort() + originalURI;
HttpUriRequest relayedRequest = new HttpGet(relayURI);
if (headerArray.length > 0) {
relayedRequest.setHeaders(headerArray);
relayURI = protocol + "://" + host + ":" + port + originalURI;
if (request.getMethod().equals(HttpMethod.HEAD)) {
HttpHead relayedRequest = new HttpHead(relayURI);
if (headerArray.length > 0) {
relayedRequest.setHeaders(headerArray);
}
relayedResponse = client.execute(relayedRequest);
} else if (request.getMethod().equals(HttpMethod.GET)) {
HttpGet relayedRequest = new HttpGet(relayURI);
if (headerArray.length > 0) {
relayedRequest.setHeaders(headerArray);
}
relayedResponse = client.execute(relayedRequest);
} else if (request.getMethod().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);
} else if (request.getMethod().equals(HttpMethod.PUT)) {
HttpPut relayedRequest = new HttpPut(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);
} else if (request.getMethod().equals(HttpMethod.DELETE)) {
HttpDelete relayedRequest = new HttpDelete(relayURI);
if (headerArray.length > 0) {
relayedRequest.setHeaders(headerArray);
}
relayedResponse = client.execute(relayedRequest);
} else if (request.getMethod().equals(HttpMethod.OPTIONS)) {
HttpOptions relayedRequest = new HttpOptions(relayURI);
if (headerArray.length > 0) {
relayedRequest.setHeaders(headerArray);
}
relayedResponse = client.execute(relayedRequest);
}
relayedResponse = client.execute(relayedRequest);
} catch (Exception e) {
String message = e.getMessage();
if (message == null) {
@@ -129,16 +174,22 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
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()));
LOG.debug("RequestURL:{} ResponseCode:{}", relayURI, relayedResponse.getStatusLine().getStatusCode());
HttpEntity httpEntity = relayedResponse.getEntity();
if (httpEntity == null) {
response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.valueOf(relayedResponse.getStatusLine().getStatusCode()));
} else {
ByteArrayOutputStream baos = new MyByteArrayOutputStream();
StreamUtils.copy(httpEntity.getContent(), baos);
response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.valueOf(relayedResponse.getStatusLine().getStatusCode()),
Unpooled.copiedBuffer(baos.toByteArray()));
response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
}
for (Header h : relayedResponse.getAllHeaders()) {
response.headers().add(h.getName(), h.getValue());
}
response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
sendResponse(ctx, response);
}
} finally {
@@ -153,10 +204,6 @@ public class GrafanaRelayHandler extends SimpleChannelInboundHandler<HttpRequest
}
}
private boolean userHasAccount(String username) {
return false;
}
static public class MyByteArrayOutputStream extends ByteArrayOutputStream {
public MyByteArrayOutputStream() {
@@ -9,8 +9,6 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders.Names;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.cookie.ServerCookieDecoder;
import org.apache.commons.lang3.StringUtils;
@@ -84,16 +82,7 @@ public class GrafanaRequestDecoder extends MessageToMessageDecoder<FullHttpReque
HttpRequest request;
try {
if (msg.getMethod().equals(HttpMethod.GET)) {
request = new GrafanaRequest();
} else {
TimelyException e = new TimelyException(HttpResponseStatus.METHOD_NOT_ALLOWED.code(),
"unhandled method type", "");
e.addResponseHeader(Names.ALLOW, HttpMethod.GET.name() + "," + HttpMethod.POST.name());
LOG.warn("Unhandled HTTP request type {}", msg.getMethod());
throw e;
}
request = new GrafanaRequest();
Multimap<String, String> headers = HttpHeaderUtils.toMultimap(msg.headers());
((AuthenticatedRequest) request).addHeaders(headers);
X509Certificate clientCert = AuthenticationService.getClientCertificate(ctx);
@@ -5,9 +5,10 @@ import io.netty.handler.codec.http.QueryStringDecoder;
import timely.api.annotation.Http;
import timely.api.request.AuthenticatedRequest;
import timely.api.request.HttpGetRequest;
import timely.api.request.HttpPostRequest;
@Http(path = "/")
public class GrafanaRequest extends AuthenticatedRequest implements HttpGetRequest {
public class GrafanaRequest extends AuthenticatedRequest implements HttpGetRequest, HttpPostRequest {
private FullHttpRequest httpRequest = null;
@@ -16,6 +17,11 @@ public class GrafanaRequest extends AuthenticatedRequest implements HttpGetReque
return new GrafanaRequest();
}
@Override
public HttpPostRequest parseBody(String content) throws Exception {
return new GrafanaRequest();
}
public void setHttpRequest(FullHttpRequest httpRequest) {
this.httpRequest = httpRequest;
}
-15
View File
@@ -1,15 +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">
<parent>
<artifactId>timely</artifactId>
<groupId>timely</groupId>
<version>1.1.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>grafana-authentication</artifactId>
</project>
@@ -1,576 +0,0 @@
package timely.balancer;
import static timely.Server.SERVICE_DISCOVERY_PATH;
import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
import java.io.File;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.net.ssl.SSLContext;
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.SslProvider;
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 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.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import timely.auth.AuthCache;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.configuration.ClientSsl;
import timely.balancer.configuration.SpringBootstrap;
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.healthcheck.HealthChecker;
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.balancer.resolver.BalancedMetricResolver;
import timely.balancer.resolver.MetricResolver;
import timely.client.http.HttpClient;
import timely.configuration.ServerSsl;
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;
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 int EPOLL_MIN_MAJOR_VERSION = 2;
private static final int EPOLL_MIN_MINOR_VERSION = 6;
private static final int EPOLL_MIN_PATCH_VERSION = 32;
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);
static ConfigurableApplicationContext applicationContext;
private final BalancerConfiguration balancerConfig;
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<>();
private CuratorFramework curatorFramework;
protected volatile boolean shutdown = false;
private static final 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 };
static {
DEFAULT_EVENT_LOOP_THREADS = Math.max(1,
SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
}
public Balancer(BalancerConfiguration balancerConf) throws Exception {
this.balancerConfig = balancerConf;
}
protected static ConfigurableApplicationContext initializeConfiguration(String[] args) {
return new SpringApplicationBuilder(SpringBootstrap.class).web(false).run(args);
}
private void shutdownHook() {
final Runnable shutdownRunner = () -> {
if (!shutdown) {
shutdown();
}
};
final Thread hook = new Thread(shutdownRunner, "shutdown-hook-thread");
Runtime.getRuntime().addShutdownHook(hook);
}
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());
}
for (Channel c : udpChannelHandleList) {
LOG.info("Closing udpChannelHandle");
channelFutures.add(c.close());
}
// wait for the channels to shutdown
channelFutures.forEach(f -> {
try {
f.get();
} catch (final Exception e) {
LOG.error("Error while shutting down channel: {}", f.channel().config());
LOG.error("{}", e.getMessage());
e.printStackTrace();
}
});
Integer quietPeriod = balancerConfig.getServer().getShutdownQuietPeriod();
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("Error while shutting down group: {}", f.toString());
LOG.error("{}", e.getMessage());
e.printStackTrace();
}
});
try {
LOG.info("Closing WebSocketRequestDecoder");
WebSocketRequestDecoder.close();
} catch (Exception e) {
LOG.error("Error closing WebSocketRequestDecoder during shutdown", e);
}
if (applicationContext != null) {
LOG.info("Closing applicationContext");
applicationContext.close();
}
if (curatorFramework != null) {
curatorFramework.close();
}
this.shutdown = true;
LOG.info("Server shut down.");
}
private static boolean useEpoll() {
// Should we just return true if this is Linux and if we get an error
// during Epoll
// setup handle it there?
final String os = SystemPropertyUtil.get(OS_NAME).toLowerCase().trim();
final String[] version = SystemPropertyUtil.get(OS_VERSION).toLowerCase().trim().split("\\.");
if (os.startsWith("linux") && version.length >= 3) {
final int major = Integer.parseInt(version[0]);
if (major > EPOLL_MIN_MAJOR_VERSION) {
return true;
} else if (major == EPOLL_MIN_MAJOR_VERSION) {
final int minor = Integer.parseInt(version[1]);
if (minor > EPOLL_MIN_MINOR_VERSION) {
return true;
} else if (minor == EPOLL_MIN_MINOR_VERSION) {
final int patch = Integer.parseInt(version[2].substring(0, 2));
return patch >= EPOLL_MIN_PATCH_VERSION;
} else {
return false;
}
} else {
return false;
}
} else {
return false;
}
}
protected SslContext createSSLContext(BalancerConfiguration config) throws Exception {
ServerSsl sslCfg = config.getSecurity().getServerSsl();
Boolean generate = sslCfg.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 {
String cert = sslCfg.getCertificateFile();
String key = sslCfg.getKeyFile();
String keyPass = sslCfg.getKeyPassword();
if (null == cert || null == key) {
throw new IllegalArgumentException("Check your SSL properties, something is wrong.");
}
ssl = SslContextBuilder.forServer(new File(cert), new File(key), keyPass);
}
ssl.ciphers(sslCfg.getUseCiphers());
// Can't set to REQUIRE because the CORS pre-flight requests will fail.
ssl.clientAuth(ClientAuth.OPTIONAL);
Boolean useOpenSSL = sslCfg.isUseOpenssl();
if (useOpenSSL) {
ssl.sslProvider(SslProvider.OPENSSL);
} else {
ssl.sslProvider(SslProvider.JDK);
}
String trustStore = sslCfg.getTrustStoreFile();
if (null != trustStore) {
if (!trustStore.isEmpty()) {
ssl.trustManager(new File(trustStore));
}
}
return ssl.build();
}
protected SSLContext createSSLClientContext(BalancerConfiguration config) throws Exception {
ClientSsl ssl = config.getSecurity().getClientSsl();
return HttpClient.getSSLContext(ssl.getTrustStoreFile(), ssl.getTrustStoreType(), ssl.getTrustStorePassword(),
ssl.getKeyFile(), ssl.getKeyType(), ssl.getKeyPassword());
}
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());
}
}
}
public void run() throws Exception {
// start curator framework
RetryPolicy retryPolicy = new RetryForever(1000);
curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
retryPolicy);
curatorFramework.start();
ensureZkPaths(curatorFramework, zkPaths);
AuthCache.setSessionMaxAge(balancerConfig.getSecurity());
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 < balancerConfig.getServer().getNumTcpPools(); x++) {
tcpClientPools.add(new TcpClientPool(this.balancerConfig));
}
HealthChecker healthChecker = new HealthChecker(this.balancerConfig, tcpClientPools.get(0));
this.metricResolver = new BalancedMetricResolver(curatorFramework, this.balancerConfig, healthChecker);
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.option(ChannelOption.SO_KEEPALIVE, true);
final int tcpPort = balancerConfig.getServer().getTcpPort();
final String tcpIp = balancerConfig.getServer().getIp();
tcpChannelHandle = tcpServer.bind(tcpIp, tcpPort).sync().channel();
final String tcpAddress = ((InetSocketAddress) tcpChannelHandle.localAddress()).getAddress().getHostAddress();
final int httpPort = balancerConfig.getHttp().getPort();
final String httpIp = balancerConfig.getHttp().getIp();
SslContext sslCtx = createSSLContext(balancerConfig);
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(balancerConfig.getSecurity().getSessionMaxAge());
}
SSLContext clientSSLContext = createSSLClientContext(balancerConfig);
final ServerBootstrap httpServer = new ServerBootstrap();
httpServer.group(httpBossGroup, httpWorkerGroup);
httpServer.channel(channelClass);
httpServer.handler(new LoggingHandler());
HttpClientPool httpClientPool = new HttpClientPool(balancerConfig, clientSSLContext);
httpServer.childHandler(setupHttpChannel(balancerConfig, sslCtx, metricResolver, httpClientPool));
httpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
httpServer.option(ChannelOption.SO_BACKLOG, 128);
httpServer.option(ChannelOption.SO_KEEPALIVE, true);
httpChannelHandle = httpServer.bind(httpIp, httpPort).sync().channel();
final String httpAddress = ((InetSocketAddress) httpChannelHandle.localAddress()).getAddress().getHostAddress();
final int wsPort = balancerConfig.getWebsocket().getPort();
final String wsIp = balancerConfig.getWebsocket().getIp();
final ServerBootstrap wsServer = new ServerBootstrap();
wsServer.group(wsBossGroup, wsWorkerGroup);
wsServer.channel(channelClass);
wsServer.handler(new LoggingHandler());
WsClientPool wsClientPool = new WsClientPool(balancerConfig, clientSSLContext);
wsServer.childHandler(setupWSChannel(balancerConfig, sslCtx, metricResolver, wsClientPool));
wsServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
wsServer.option(ChannelOption.SO_BACKLOG, 128);
wsServer.option(ChannelOption.SO_KEEPALIVE, true);
/* Not sure if next three lines are necessary */
wsServer.option(ChannelOption.SO_SNDBUF, 1048576);
wsServer.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 838860);
wsServer.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 620145);
wsChannelHandle = wsServer.bind(wsIp, wsPort).sync().channel();
final String wsAddress = ((InetSocketAddress) wsChannelHandle.localAddress()).getAddress().getHostAddress();
final int udpPort = balancerConfig.getServer().getUdpPort();
final String udpIp = balancerConfig.getServer().getIp();
UdpClientPool udpClientPool = new UdpClientPool(balancerConfig);
for (int n = 1; 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(udpServer.bind(udpIp, udpPort).sync().channel());
}
shutdownHook();
LOG.info(
"Server started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, {}:{} for WebSocket traffic, and {}:{} for UDP traffic",
tcpAddress, tcpPort, httpAddress, httpPort, wsAddress, wsPort, wsAddress, udpPort);
}
public static void main(String[] args) throws Exception {
Balancer.applicationContext = Balancer.initializeConfiguration(args);
BalancerConfiguration balancerConf = applicationContext.getBean(BalancerConfiguration.class);
Balancer b = new Balancer(balancerConf);
try {
b.run();
LATCH.await();
} catch (final InterruptedException e) {
LOG.info("Server shutting down.");
} catch (Exception e) {
LOG.error("Error running server.", e);
} finally {
try {
b.shutdown();
} catch (Exception e) {
System.exit(1);
}
}
}
protected ChannelHandler setupHttpChannel(BalancerConfiguration balancerConfig, SslContext sslCtx,
MetricResolver metricResolver, HttpClientPool httpClientPool) {
return new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerConfig.getHttp(), sslCtx));
ch.pipeline().addLast("encoder", new HttpResponseEncoder());
ch.pipeline().addLast("decoder", new HttpRequestDecoder());
ch.pipeline().addLast("compressor", new HttpContentCompressor());
ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
ch.pipeline().addLast("queryDecoder", new timely.netty.http.HttpRequestDecoder(
balancerConfig.getSecurity(), balancerConfig.getHttp()));
ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler());
ch.pipeline().addLast("login",
new X509LoginRequestHandler(balancerConfig.getSecurity(), balancerConfig.getHttp()));
ch.pipeline().addLast("httpRelay",
new HttpRelayHandler(balancerConfig, metricResolver, httpClientPool));
ch.pipeline().addLast("error", new TimelyExceptionHandler());
}
};
}
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(8192, 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(8192, 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(BalancerConfiguration balancerConfig, 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(8192));
ch.pipeline().addLast("sessionExtractor", new WebSocketFullRequestHandler());
ch.pipeline().addLast("idle-handler",
new IdleStateHandler(balancerConfig.getWebsocket().getTimeout(), 0, 0));
ch.pipeline().addLast("ws-protocol",
new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, true));
ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(balancerConfig.getSecurity()));
ch.pipeline().addLast("httpRelay", new WsRelayHandler(balancerConfig, metricResolver, wsClientPool));
ch.pipeline().addLast("error", new WSTimelyExceptionHandler());
}
};
}
}
@@ -1,121 +0,0 @@
package timely.balancer.configuration;
import javax.validation.Valid;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.stereotype.Component;
import timely.configuration.Cache;
@Component
@ConfigurationProperties(prefix = "timely-balancer")
public class BalancerConfiguration {
@Valid
@NestedConfigurationProperty
private BalancerServer server = new BalancerServer();
@Valid
@NestedConfigurationProperty
private BalancerHttp http = new BalancerHttp();
@Valid
@NestedConfigurationProperty
private BalancerWebsocket websocket = new BalancerWebsocket();
@Valid
@NestedConfigurationProperty
private ZooKeeper zooKeeper = new ZooKeeper();
@Valid
@NestedConfigurationProperty
private BalancerSecurity security = new BalancerSecurity();
@Valid
@NestedConfigurationProperty
private Cache cache = new Cache();
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 ZooKeeper getZooKeeper() {
return zooKeeper;
}
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 BalancerSecurity getSecurity() {
return security;
}
public BalancerServer getServer() {
return server;
}
public BalancerHttp getHttp() {
return http;
}
public BalancerWebsocket getWebsocket() {
return websocket;
}
public Cache getCache() {
return cache;
}
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,7 +0,0 @@
package timely.balancer.configuration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication(scanBasePackageClasses = { SpringBootstrap.class })
public class SpringBootstrap {
}
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>timely-app</artifactId>
<packaging>pom</packaging>
+2 -2
View File
@@ -23,8 +23,8 @@
{"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
{"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
],
"version": "1.1.0",
"updated": "2019-01-15"
"version": "1.2.4",
"updated": "2019-05-22"
},
"includes": [
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "timely-app",
"private": true,
"version": "1.1.0",
"version": "1.2.4",
"description": "",
"main": "index.js",
"scripts": {
+2 -2
View File
@@ -23,8 +23,8 @@
{"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
{"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
],
"version": "1.1.0",
"updated": "2019-01-15"
"version": "1.2.4",
"updated": "2019-05-22"
},
"includes": [
+27 -23
View File
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
<packaging>pom</packaging>
<name>Timely</name>
<description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -43,31 +43,29 @@
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<version.accumulo>1.9.2</version.accumulo>
<version.buildnumber-maven>1.4</version.buildnumber-maven>
<version.caffeine>2.3.3</version.caffeine>
<version.caffeine>2.6.2</version.caffeine>
<version.collectd>5.7.1</version.collectd>
<version.commons-collections4>4.1</version.commons-collections4>
<version.commons-configuration>1.10</version.commons-configuration>
<version.commons-io>2.6</version.commons-io>
<version.commons-lang3>3.5</version.commons-lang3>
<version.commons-pool2>2.6.0</version.commons-pool2>
<version.curator>2.11.1</version.curator>
<version.disruptor>3.3.6</version.disruptor>
<version.commons-collections4>4.3</version.commons-collections4>
<version.commons-lang3>3.8.1</version.commons-lang3>
<version.commons-pool2>2.6.1</version.commons-pool2>
<version.curator>4.1.0</version.curator>
<version.disruptor>3.4.2</version.disruptor>
<version.eclipse-lifecycle>1.0.0</version.eclipse-lifecycle>
<version.fast-classpath-scanner>2.0.5</version.fast-classpath-scanner>
<version.fast-classpath-scanner>3.1.13</version.fast-classpath-scanner>
<version.flink>1.1.4</version.flink>
<version.google-flatbuffers>1.10.0</version.google-flatbuffers>
<version.guava>19.0</version.guava>
<version.guava>27.1-jre</version.guava>
<version.hadoop>2.7.3</version.hadoop>
<version.httpcomponents>4.5.2</version.httpcomponents>
<version.httpcomponents>4.5.6</version.httpcomponents>
<version.impsort>1.2.0</version.impsort>
<version.jackson>2.9.8</version.jackson>
<version.jacoco>0.7.6.201602180812</version.jacoco>
<version.javacsv>2.0</version.javacsv>
<version.javassist>3.21.0-GA</version.javassist>
<version.javassist>3.24.1-GA</version.javassist>
<version.jcip-annotations>1.0</version.jcip-annotations>
<version.jsoup>1.9.2</version.jsoup>
<version.jsoup>1.11.3</version.jsoup>
<version.junit>4.12</version.junit>
<version.log4j>2.6.2</version.log4j>
<version.log4j>2.11.2</version.log4j>
<version.maven-assembly>2.6</version.maven-assembly>
<version.maven-build-helper>1.12</version.maven-build-helper>
<version.maven-compiler>3.3</version.maven-compiler>
@@ -77,14 +75,14 @@
<version.maven-rpm>2.1.4</version.maven-rpm>
<version.maven-shade>2.4.2</version.maven-shade>
<version.maven-surefire>2.19.1</version.maven-surefire>
<version.netty>4.0.56.Final</version.netty>
<version.netty-tcnative>2.0.7.Final</version.netty-tcnative>
<version.netty>4.1.35.Final</version.netty>
<version.netty-tcnative>2.0.25.Final</version.netty-tcnative>
<version.os-maven>1.5.0.Final</version.os-maven>
<version.powermock>1.6.4</version.powermock>
<version.slf4j>1.7.21</version.slf4j>
<version.slf4j>1.7.26</version.slf4j>
<version.spotbugs-maven>3.1.7</version.spotbugs-maven>
<version.spring-boot>1.4.1.RELEASE</version.spring-boot>
<version.tyrus>1.13</version.tyrus>
<version.spring-boot>2.1.3.RELEASE</version.spring-boot>
<version.tyrus>1.15</version.tyrus>
</properties>
<dependencyManagement>
<dependencies>
@@ -141,7 +139,7 @@
<dependency>
<groupId>io.github.lukehutch</groupId>
<artifactId>fast-classpath-scanner</artifactId>
<version>2.0.5</version>
<version>${version.fast-classpath-scanner}</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
@@ -157,6 +155,12 @@
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-core</artifactId>
<version>${version.accumulo}</version>
<exclusions>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.accumulo</groupId>
@@ -452,7 +456,7 @@
<artifactId>maven-failsafe-plugin</artifactId>
<version>${version.maven-surefire}</version>
<configuration>
<argLine>${failsafeArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative.so -Dzookeeper.sasl.client=false</argLine>
<argLine>${failsafeArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty_tcnative_linux_x86_64.so -Dzookeeper.sasl.client=false</argLine>
<includes>
<include>**/*.java</include>
</includes>
@@ -495,7 +499,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>${version.maven-surefire}</version>
<configuration>
<argLine>${sureFireArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative.so -Dzookeeper.sasl.client=false</argLine>
<argLine>${sureFireArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative_linux_x86_64.so -Dzookeeper.sasl.client=false</argLine>
<excludedGroups>timely.test.IntegrationTest</excludedGroups>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
</configuration>
+4 -5
View File
@@ -27,15 +27,14 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
$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}/*"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
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} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely"
$JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely
+6 -10
View File
@@ -2,17 +2,12 @@
if [[ `uname` == "Darwin" ]]; then
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
TCNATIVE_SUFFIX="tcnative.jnilib"
TCNATIVE_SUFFIX="jnilib"
else
THIS_SCRIPT=`readlink -f $0`
TCNATIVE_SUFFIX="tcnative.so"
TCNATIVE_SUFFIX="so"
fi
# netty tcnative file reference
#netty-tcnative-1.1.33.Fork18-linux-x86_64-fedora.jar -> libnetty-tcnative.so
#netty-tcnative-1.1.33.Fork18-linux-x86_64.jar -> libnetty-tcnative.so
#netty-tcnative-1.1.33.Fork18-osx-x86_64.jar -> libnetty-tcnative.jnilib
THIS_DIR="${THIS_SCRIPT%/*}"
NATIVE_DIR="${THIS_DIR}/META-INF/native"
BASE_DIR=${THIS_DIR}/..
@@ -32,11 +27,12 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty-${TCNATIVE_SUFFIX}
$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}/*"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
# for debugging
+5 -2
View File
@@ -46,9 +46,12 @@
<user-service id="usernamePasswordService">
<user name="test" password="test1" authorities="A,B,C" />
</user-service>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:bean id="noOpPasswordEncoder" class="org.springframework.security.crypto.password.NoOpPasswordEncoder"/>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:property name="userDetailsService" ref="usernamePasswordService"/>
<beans:property name="passwordEncoder" ref="noOpPasswordEncoder"/>
</beans:bean>
<authentication-manager alias="authenticationManager">
+5 -3
View File
@@ -18,6 +18,8 @@ timely:
max-downsample-memory: 10000000
security:
session-max-age: 86400
cache-expiration-minutes: 60
cache-refresh-minutes: 5
allow-anonymous-http-access: true
allow-anonymous-ws-access: true
ssl:
@@ -50,9 +52,9 @@ timely:
timeout: 60
subscription-lag: 120
meta-cache:
expiration-minutes: 60
initial-capacity: 2000
max-capacity: 10000
expiration-minutes: 1440
cache-refresh-minutes: 60
max-tag-values: 100
visibility-cache:
expiration-minutes: 60
initial-capacity: 2000
+5 -3
View File
@@ -18,6 +18,8 @@ timely:
max-downsample-memory: 10000000
security:
session-max-age: 86400
cache-expiration-minutes: 60
cache-refresh-minutes: 5
allow-anonymous-http-access: true
allow-anonymous-ws-access: true
ssl:
@@ -54,9 +56,9 @@ timely:
scanner-read-ahead: 1
subscription-batch-size: 5000
meta-cache:
expiration-minutes: 60
initial-capacity: 2000
max-capacity: 10000
expiration-minutes: 1440
cache-refresh-minutes: 60
max-tag-values: 100
visibility-cache:
expiration-minutes: 60
initial-capacity: 2000
+7 -5
View File
@@ -8,6 +8,7 @@ timely:
age-off-hours:
default: 26
metrics-report-ignored-tags:
defaultVisibility:
accumulo:
instance-name: accumulo
zookeepers: localhost:2181
@@ -22,6 +23,8 @@ timely:
max-downsample-memory: 10000000
security:
session-max-age: 86400
cache-expiration-minutes: 60
cache-refresh-minutes: 5
allow-anonymous-http-access: true
allow-anonymous-ws-access: true
serverSsl:
@@ -59,10 +62,9 @@ timely:
scanner-read-ahead: 1
subscription-batch-size: 5000
meta-cache:
expiration-minutes: 60
initial-capacity: 2000
max-capacity: 10000
expiration-minutes: 1440
cache-refresh-minutes: 60
max-tag-values: 100
visibility-cache:
expiration-minutes: 60
initial-capacity: 2000
max-capacity: 10000
+3 -38
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.1.1-SNAPSHOT</version>
<version>1.2.4</version>
</parent>
<artifactId>timely-server</artifactId>
<dependencies>
@@ -149,9 +149,8 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>runtime</scope>
</dependency>
<dependency>
@@ -240,9 +239,8 @@
<artifactItems>
<artifactItem>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<type>jar</type>
<overWrite>false</overWrite>
<outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -284,37 +282,4 @@
</extension>
</extensions>
</build>
<profiles>
<profile>
<id>native-epoll-linux</id>
<activation>
<os>
<family>!mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
<classifier>linux-x86_64</classifier>
</dependency>
</dependencies>
</profile>
<profile>
<id>native-epoll</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
+8 -7
View File
@@ -70,6 +70,7 @@ import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import timely.api.response.TimelyException;
@@ -250,7 +251,7 @@ public class Server {
}
protected static Configuration initializeConfiguration(String[] args) {
applicationContext = new SpringApplicationBuilder(SpringBootstrap.class).web(false).run(args);
applicationContext = new SpringApplicationBuilder(SpringBootstrap.class).web(WebApplicationType.NONE).run(args);
return applicationContext.getBean(Configuration.class);
}
@@ -430,7 +431,7 @@ public class Server {
// Initialize the MetaCache
MetaCacheFactory.getCache(config);
// initialize the auth cache
AuthCache.setSessionMaxAge(config.getSecurity());
AuthCache.configure(config.getSecurity());
// Initialize the VisibilityCache
VisibilityCache.init(config);
final boolean useEpoll = useEpoll();
@@ -587,7 +588,7 @@ public class Server {
ch.pipeline().addLast("decoder", new HttpRequestDecoder());
ch.pipeline().addLast("compressor", new HttpContentCompressor());
ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
final Cors corsCfg = config.getHttp().getCors();
final CorsConfig.Builder ccb;
@@ -635,7 +636,7 @@ public class Server {
ch.pipeline().addLast("logger", new LoggingHandler());
ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
ch.pipeline().addLast("putDecoder", new UdpDecoder());
ch.pipeline().addLast("putHandler", new TcpPutHandler(dataStore));
}
@@ -648,7 +649,7 @@ public class Server {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
ch.pipeline().addLast("putDecoder", new TcpDecoder());
ch.pipeline().addLast("putHandler", new TcpPutHandler(dataStore));
ch.pipeline().addLast("versionHandler", new TcpVersionHandler());
@@ -663,11 +664,11 @@ public class Server {
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(8192));
ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
ch.pipeline().addLast("sessionExtractor", new WebSocketFullRequestHandler());
ch.pipeline().addLast("idle-handler", new IdleStateHandler(conf.getWebsocket().getTimeout(), 0, 0));
ch.pipeline().addLast("ws-protocol",
new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, true));
new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, false, true));
ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(config.getSecurity()));
ch.pipeline().addLast("aggregators", new WSAggregatorsRequestHandler());
ch.pipeline().addLast("metrics", new WSMetricsRequestHandler(config));
@@ -0,0 +1,53 @@
package timely.adapter.accumulo;
import org.apache.accumulo.core.client.lexicoder.LongLexicoder;
import org.apache.accumulo.core.client.lexicoder.PairLexicoder;
import org.apache.accumulo.core.client.lexicoder.StringLexicoder;
import org.apache.accumulo.core.data.Key;
import timely.api.model.Meta;
import timely.model.Metric;
import timely.model.parse.TagListParser;
import timely.model.parse.TagParser;
/**
* Creates Accumulo data structures from {@link Metric}
*/
public class MetaAdapter {
private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
new LongLexicoder());
private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
new StringLexicoder());
private static final TagParser tagParser = new TagParser();
private static final TagListParser tagListParser = new TagListParser();
public static Key createMetricKey(String metricName, Long timestamp) {
return new Key(Meta.METRIC_PREFIX + metricName, "", "", timestamp);
}
public static Key createTagKey(String metricName, String tagName, Long timestamp) {
return new Key(Meta.TAG_PREFIX + metricName, tagName, "", timestamp);
}
public static Key createValueKey(String metricName, String tagName, String tagValue, Long timestamp) {
return new Key(Meta.VALUE_PREFIX + metricName, tagName, tagValue, timestamp);
}
public static Meta parse(Key key) {
String row = key.getRow().toString();
if (row.startsWith(Meta.METRIC_PREFIX)) {
return Meta.parse(key, null, Meta.METRIC_PREFIX);
} else if (row.startsWith(Meta.TAG_PREFIX)) {
return Meta.parse(key, null, Meta.TAG_PREFIX);
} else if (row.startsWith(Meta.VALUE_PREFIX)) {
return Meta.parse(key, null, Meta.VALUE_PREFIX);
}
throw new IllegalStateException("Invalid key in meta " + key.toString());
}
public static String decodeRowKey(Key key) {
return parse(key).getMetric();
}
}
@@ -55,16 +55,43 @@ public class Meta implements Comparable<Meta> {
return keys;
}
public static Meta parse(Key k, Value v) {
public static String prefix(Key key) {
String row = key.getRow().toString();
if (row.startsWith(Meta.METRIC_PREFIX)) {
return Meta.METRIC_PREFIX;
} else if (row.startsWith(Meta.TAG_PREFIX)) {
return Meta.TAG_PREFIX;
} else if (row.startsWith(Meta.VALUE_PREFIX)) {
return Meta.VALUE_PREFIX;
}
throw new IllegalStateException("Invalid key in meta " + key.toString());
}
public static Meta parse(Key key) {
return parse(key, null);
}
public static Meta parse(Key key, Value value) {
String row = key.getRow().toString();
if (row.startsWith(Meta.METRIC_PREFIX)) {
return Meta.parse(key, value, Meta.METRIC_PREFIX);
} else if (row.startsWith(Meta.TAG_PREFIX)) {
return Meta.parse(key, value, Meta.TAG_PREFIX);
} else if (row.startsWith(Meta.VALUE_PREFIX)) {
return Meta.parse(key, value, Meta.VALUE_PREFIX);
}
throw new IllegalStateException("Invalid key in meta " + key.toString());
}
public static Meta parse(Key k, Value v, String prefix) {
if (k.getColumnQualifier().getLength() > 0) {
return new Meta(k.getRow().toString().substring(METRIC_PREFIX.length()), k.getColumnFamily().toString(),
return new Meta(k.getRow().toString().substring(prefix.length()), k.getColumnFamily().toString(),
k.getColumnQualifier().toString());
} else if (k.getColumnFamily().getLength() > 0) {
return new Meta(k.getRow().toString().substring(METRIC_PREFIX.length()), k.getColumnFamily().toString(),
null);
return new Meta(k.getRow().toString().substring(prefix.length()), k.getColumnFamily().toString(), null);
} else {
return new Meta(k.getRow().toString().substring(METRIC_PREFIX.length()), null, null);
return new Meta(k.getRow().toString().substring(prefix.length()), null, null);
}
}
@@ -23,32 +23,67 @@ public class AuthCache {
private static final Logger LOG = LoggerFactory.getLogger(AuthCache.class);
private static Cache<String, TimelyPrincipal> CACHE = null;
private static int sessionMaxAge = -1;
private static int cacheExpirationMinutes = -1;
private static int cacheRefreshMinutes = -1;
/**
* For tests only
*/
public static void resetSessionMaxAge() {
sessionMaxAge = -1;
public static void resetConfiguration() {
cacheExpirationMinutes = -1;
cacheRefreshMinutes = -1;
}
public static void setSessionMaxAge(Security security) {
if (-1 != sessionMaxAge) {
throw new IllegalStateException("Cache session max age already configured.");
public static void configure(Security security) {
if (-1 != cacheExpirationMinutes || -1 != cacheRefreshMinutes) {
throw new IllegalStateException("Cache already configured.");
}
sessionMaxAge = security.getSessionMaxAge();
cacheExpirationMinutes = security.getCacheExpirationMinutes();
cacheRefreshMinutes = security.getCacheRefreshMinutes();
}
private static Cache<String, TimelyPrincipal> getCache() {
if (-1 == sessionMaxAge) {
if (-1 == cacheExpirationMinutes) {
throw new IllegalStateException("Cache session max age not configured.");
}
if (null == CACHE) {
CACHE = Caffeine.newBuilder().expireAfterAccess(sessionMaxAge, TimeUnit.SECONDS).build();
Caffeine<Object, Object> caffeine = Caffeine.newBuilder();
caffeine.expireAfterWrite(cacheExpirationMinutes, TimeUnit.MINUTES);
if (cacheRefreshMinutes > 0) {
caffeine.refreshAfterWrite(cacheRefreshMinutes, TimeUnit.MINUTES);
CACHE = caffeine.build(key -> getTimelyPrincipal(key));
} else {
CACHE = caffeine.build();
}
}
return CACHE;
}
private static TimelyPrincipal getTimelyPrincipal(String key) {
LOG.debug("Refreshing TimelyPrincipal {}", key);
TimelyPrincipal timelyPrincipal = null;
String[] dnArray = key.split(" -> ");
if (dnArray.length == 1) {
SubjectIssuerDNPair pair = SubjectIssuerDNPair.parse(dnArray[0]);
TimelyAuthenticationToken token = AuthenticationService.getAuthenticationToken(new Object(),
pair.subjectDN(), pair.issuerDN());
timelyPrincipal = AuthenticationService.authenticate(token, pair, token.getTimelyPrincipal().getName(),
false);
} else if (dnArray.length > 1) {
Collection<TimelyUser> timelyUsers = new ArrayList<>();
for (String s : dnArray) {
SubjectIssuerDNPair pair = SubjectIssuerDNPair.parse(s);
TimelyAuthenticationToken token = AuthenticationService.getAuthenticationToken(new Object(),
pair.subjectDN(), pair.issuerDN());
TimelyPrincipal p = AuthenticationService.authenticate(token, pair,
token.getTimelyPrincipal().getName(), false);
timelyUsers.addAll(p.getProxiedUsers());
}
timelyPrincipal = new TimelyPrincipal(timelyUsers);
}
return timelyPrincipal;
}
public static boolean containsKey(String key) {
return getCache().getIfPresent(key) != null;
}
@@ -23,7 +23,6 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.security.authentication.AuthenticationManager;
import org.springframework.security.core.Authentication;
import org.springframework.security.web.authentication.preauth.x509.SubjectDnX509PrincipalExtractor;
import timely.api.request.AuthenticatedRequest;
import timely.api.response.TimelyException;
import timely.auth.util.DnUtils;
@@ -35,7 +34,6 @@ public class AuthenticationService {
private static final Logger LOG = LoggerFactory.getLogger(AuthenticationService.class);
private static ApplicationContext springContext = null;
private static AuthenticationManager authManager = null;
private static SubjectDnX509PrincipalExtractor x509 = null;
private static ConcurrentHashMap<String, Object> fetchingEntity = new ConcurrentHashMap<>();
private static ConcurrentHashMap<String, Object> fetchingPrincipal = new ConcurrentHashMap<>();
private static Collection<String> requiredRoles;
@@ -45,7 +43,6 @@ public class AuthenticationService {
try {
springContext = new ClassPathXmlApplicationContext("security.xml");
authManager = springContext.getBean("authenticationManager", AuthenticationManager.class);
x509 = springContext.getBean("x509PrincipalExtractor", SubjectDnX509PrincipalExtractor.class);
requiredRoles = springContext.getBean("requiredRoles", List.class);
requiredRoles.removeIf(String::isEmpty);
requiredAuths = springContext.getBean("requiredAuths", List.class);
@@ -65,9 +62,17 @@ public class AuthenticationService {
}
public static TimelyPrincipal authenticate(Authentication authentication, SubjectIssuerDNPair pair, String entity) {
return authenticate(authentication, pair, entity, true);
}
public static TimelyPrincipal authenticate(Authentication authentication, SubjectIssuerDNPair pair, String entity,
boolean useCache) {
// only one thread should call the authenticationManager for a given entity
TimelyPrincipal principal = AuthCache.get(entity);
TimelyPrincipal principal = null;
if (useCache) {
principal = AuthCache.get(entity);
}
// once an entity is in the AuthCache, we can bypass all of this
if (principal == null) {
Object newObj = new Object();
@@ -79,12 +84,16 @@ public class AuthenticationService {
synchronized (o) {
// if multiple threads are waiting on this synchronized, then the 2nd and all
// others will get the principal here
principal = AuthCache.get(entity);
if (useCache) {
principal = AuthCache.get(entity);
}
if (principal == null) {
Authentication token = getAuthenticationManager().authenticate(authentication);
principal = getTimelyPrincipal(token, pair);
LOG.trace("Got entity principal for {} from AuthenticationManager", entity);
AuthCache.put(entity, principal);
if (useCache) {
AuthCache.put(entity, principal);
}
} else {
LOG.trace("Got entity principal for {} from AuthCache on 2nd attempt", entity);
}
@@ -232,7 +241,7 @@ public class AuthenticationService {
return authenticationToken;
}
public static TimelyAuthenticationToken getAuthenticationToken(X509Certificate clientCert, String subjectDn,
public static TimelyAuthenticationToken getAuthenticationToken(Object clientCert, String subjectDn,
String issuerDn) {
TimelyAuthenticationToken authenticationToken = null;
try {
@@ -9,7 +9,9 @@ import javax.validation.Valid;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.stereotype.Component;
import org.springframework.validation.annotation.Validated;
@Validated
@Component
@ConfigurationProperties(prefix = "timely")
public class Configuration {
@@ -19,6 +21,7 @@ public class Configuration {
private HashMap<String, Integer> metricAgeOffDays = new HashMap<>();
private List<String> metricsReportIgnoredTags = new ArrayList<>();
private String instance = null;
private String defaultVisibility = null;
@Valid
@NestedConfigurationProperty
@@ -71,6 +74,14 @@ public class Configuration {
return instance;
}
public String getDefaultVisibility() {
return defaultVisibility;
}
public void setDefaultVisibility(String defaultVisibility) {
this.defaultVisibility = defaultVisibility;
}
public HashMap<String, Integer> getMetricAgeOffDays() {
return metricAgeOffDays;
}
@@ -2,9 +2,9 @@ package timely.configuration;
public class MetaCache {
private long expirationMinutes = 60;
private int initialCapacity = 2000;
private long maxCapacity = 10000;
private long expirationMinutes = 1440;
private int cacheRefreshMinutes = -1;
private long maxTagValues = 100;
public long getExpirationMinutes() {
return expirationMinutes;
@@ -14,19 +14,19 @@ public class MetaCache {
this.expirationMinutes = expirationMinutes;
}
public int getInitialCapacity() {
return initialCapacity;
public void setCacheRefreshMinutes(int cacheRefreshMinutes) {
this.cacheRefreshMinutes = cacheRefreshMinutes;
}
public void setInitialCapacity(int initialCapacity) {
this.initialCapacity = initialCapacity;
public int getCacheRefreshMinutes() {
return cacheRefreshMinutes;
}
public long getMaxCapacity() {
return maxCapacity;
public long getMaxTagValues() {
return maxTagValues;
}
public void setMaxCapacity(long maxCapacity) {
this.maxCapacity = maxCapacity;
public void setMaxTagValues(long maxTagValues) {
this.maxTagValues = maxTagValues;
}
}
@@ -9,6 +9,8 @@ public class Security {
private boolean allowAnonymousHttpAccess = false;
private boolean allowAnonymousWsAccess = false;
private int sessionMaxAge = 86400;
private int cacheExpirationMinutes = 60;
private int cacheRefreshMinutes = 5;
@Valid
@NestedConfigurationProperty
private ServerSsl serverSsl = new ServerSsl();
@@ -37,6 +39,22 @@ public class Security {
this.sessionMaxAge = sessionMaxAge;
}
public void setCacheExpirationMinutes(int cacheExpirationMinutes) {
this.cacheExpirationMinutes = cacheExpirationMinutes;
}
public int getCacheExpirationMinutes() {
return cacheExpirationMinutes;
}
public void setCacheRefreshMinutes(int cacheRefreshMinutes) {
this.cacheRefreshMinutes = cacheRefreshMinutes;
}
public int getCacheRefreshMinutes() {
return cacheRefreshMinutes;
}
public ServerSsl getServerSsl() {
return serverSsl;
}
@@ -22,19 +22,20 @@ public class TimelyAuthenticationToken extends PreAuthenticatedAuthenticationTok
public static final String PROXIED_ISSUERS_HEADER = "X-ProxiedIssuersChain";
private Multimap<String, String> httpHeaders = null;
private X509Certificate clientCert;
private X509Certificate clientCert = null;
private TimelyPrincipal timelyPrincipal;
public TimelyAuthenticationToken(String subjectDn, X509Certificate clientCert,
Multimap<String, String> httpHeaders) {
public TimelyAuthenticationToken(String subjectDn, Object clientCert, Multimap<String, String> httpHeaders) {
super(subjectDn, clientCert);
if (httpHeaders == null) {
this.httpHeaders = HashMultimap.create();
} else {
this.httpHeaders = httpHeaders;
}
this.clientCert = clientCert;
String issuerDn = clientCert == null ? null : clientCert.getIssuerDN().getName();
if (clientCert instanceof X509Certificate) {
this.clientCert = (X509Certificate) clientCert;
}
String issuerDn = clientCert == null ? null : this.clientCert.getIssuerDN().getName();
String proxiedEntities;
String proxiedIssuers;
try {
@@ -61,9 +62,11 @@ public class TimelyAuthenticationToken extends PreAuthenticatedAuthenticationTok
LOG.trace("Created TimelyAuthenticationToken: {}", timelyPrincipal.getName());
}
public TimelyAuthenticationToken(String subjectDn, String issuerDn, X509Certificate clientCert) {
public TimelyAuthenticationToken(String subjectDn, String issuerDn, Object clientCert) {
super(subjectDn, clientCert);
this.clientCert = clientCert;
if (clientCert instanceof X509Certificate) {
this.clientCert = (X509Certificate) clientCert;
}
TimelyUser.UserType userType = DnUtils.isServerDN(subjectDn) ? TimelyUser.UserType.SERVER
: TimelyUser.UserType.USER;
TimelyUser timelyUser = new TimelyUser(SubjectIssuerDNPair.of(subjectDn, issuerDn), userType, null, null, null,
@@ -2,6 +2,7 @@ package timely.store;
import static org.apache.accumulo.core.conf.AccumuloConfiguration.getMemoryInBytes;
import static org.apache.accumulo.core.conf.AccumuloConfiguration.getTimeInMillis;
import static timely.adapter.accumulo.MetricAdapter.VISIBILITY_TAG;
import java.io.IOException;
import java.nio.charset.Charset;
@@ -18,6 +19,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.SortedMap;
import java.util.Timer;
@@ -50,8 +52,11 @@ import org.apache.accumulo.core.data.*;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.user.RegExFilter;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.security.VisibilityEvaluator;
import org.apache.accumulo.core.util.Pair;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -130,6 +135,7 @@ public class DataStoreImpl implements DataStore {
private final Map<String, String> ageOff;
private final long defaultAgeOffMilliSec;
private DataStoreCache cache = null;
private final String defaultVisibility;
public DataStoreImpl(Configuration conf, int numWriteThreads) throws TimelyException {
@@ -149,6 +155,31 @@ public class DataStoreImpl implements DataStore {
scannerThreads = accumuloConf.getScan().getThreads();
maxDownsampleMemory = accumuloConf.getScan().getMaxDownsampleMemory();
security = conf.getSecurity();
defaultVisibility = conf.getDefaultVisibility();
if (StringUtils.isNotBlank(defaultVisibility)) {
ColumnVisibility vis;
try {
// validate visibility
vis = new ColumnVisibility(defaultVisibility);
LOG.info("Using defaultVisibility: {}", vis.toString());
} catch (RuntimeException e) {
LOG.error("Error validating defaultVisibility " + defaultVisibility + " " + e.getMessage(), e);
throw e;
}
try {
String whoami = connector.whoami();
Authorizations auths = connector.securityOperations().getUserAuthorizations(whoami);
VisibilityEvaluator eval = new VisibilityEvaluator(auths);
if (!eval.evaluate(vis)) {
throw new IllegalArgumentException("Accumulo user " + whoami + " with authorization "
+ auths.toString() + " can not see data with defaultVisibility " + vis.toString());
}
} catch (RuntimeException e) {
LOG.error(e.getMessage(), e);
throw e;
}
}
metricsTable = conf.getMetricsTable();
if (metricsTable.contains(".")) {
@@ -177,7 +208,7 @@ public class DataStoreImpl implements DataStore {
}
try {
this.removeAgeOffIterators(connector, metricsTable);
this.applyAgeOffIterator(connector, metricsTable, true);
this.applyMetricAgeOffIterator(connector, metricsTable);
} catch (Exception e1) {
Throwable cause = e1.getCause();
if (cause.getMessage().contains("conflict")) {
@@ -198,7 +229,7 @@ public class DataStoreImpl implements DataStore {
}
try {
this.removeAgeOffIterators(connector, metaTable);
this.applyAgeOffIterator(connector, metaTable, false);
this.applyMetaAgeOffIterator(connector, metaTable);
} catch (Exception e1) {
Throwable cause = e1.getCause();
if (cause.getMessage().contains("conflict")) {
@@ -251,13 +282,15 @@ public class DataStoreImpl implements DataStore {
}
}
private void applyAgeOffIterator(Connector con, String tableName, boolean useIterator) throws Exception {
IteratorSetting ageOffIteratorSettings = null;
if (useIterator) {
ageOffIteratorSettings = new IteratorSetting(100, "ageoff", MetricAgeOffIterator.class, this.ageOff);
} else {
ageOffIteratorSettings = new IteratorSetting(100, "ageoff", MetricAgeOffFilter.class, this.ageOff);
}
private void applyMetricAgeOffIterator(Connector con, String tableName) throws Exception {
IteratorSetting ageOffIteratorSettings = new IteratorSetting(100, "ageoffmetrics", MetricAgeOffIterator.class,
this.ageOff);
connector.tableOperations().attachIterator(tableName, ageOffIteratorSettings, AGEOFF_SCOPES);
}
private void applyMetaAgeOffIterator(Connector con, String tableName) throws Exception {
IteratorSetting ageOffIteratorSettings = new IteratorSetting(100, "ageoffmeta", MetaAgeOffIterator.class,
this.ageOff);
connector.tableOperations().attachIterator(tableName, ageOffIteratorSettings, AGEOFF_SCOPES);
}
@@ -279,6 +312,17 @@ public class DataStoreImpl implements DataStore {
public void store(Metric metric, boolean cacheEnabled) {
LOG.trace("Received Store Request for: {}", metric);
// if default visibility is configured and current metric does not contain a
// visibility
// add it here, so that the cache and accumulo stores both have the correct
// visibility
if (StringUtils.isNotBlank(defaultVisibility)) {
Optional<Tag> visTag = metric.getTags().stream().filter(t -> t.getKey().equals(VISIBILITY_TAG)).findFirst();
if (!visTag.isPresent()) {
metric.addTag(new Tag(VISIBILITY_TAG, defaultVisibility));
}
}
if (cache != null && cacheEnabled) {
cache.store(metric);
}
@@ -396,6 +440,7 @@ public class DataStoreImpl implements DataStore {
@Override
public SuggestResponse suggest(SuggestRequest request) throws TimelyException {
SuggestResponse result = new SuggestResponse();
Scanner scanner = null;
try {
if (request.getType().equals("metrics")) {
Range range;
@@ -415,7 +460,7 @@ public class DataStoreImpl implements DataStore {
end.append(lastBytes, 0, lastBytes.length);
range = new Range(start, end);
}
Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
scanner.setRange(range);
List<String> metrics = new ArrayList<>();
for (Entry<Key, Value> metric : scanner) {
@@ -430,6 +475,10 @@ public class DataStoreImpl implements DataStore {
LOG.error("Error during suggest: " + ex.getMessage(), ex);
throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
"Error during suggest: " + ex.getMessage(), ex.getMessage(), ex);
} finally {
if (scanner != null) {
scanner.close();
}
}
return result;
}
@@ -464,28 +513,34 @@ public class DataStoreImpl implements DataStore {
tagPatterns.put(k, Pattern.compile(v));
});
try {
List<Result> resultField = new ArrayList<>();
Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
Key start = new Key(Meta.VALUE_PREFIX + msg.getQuery());
Key end = start.followingKey(PartialKey.ROW);
Range range = new Range(start, end);
scanner.setRange(range);
tags.keySet().forEach(k -> scanner.fetchColumnFamily(new Text(k)));
int total = 0;
for (Entry<Key, Value> entry : scanner) {
Meta metaEntry = Meta.parse(entry.getKey(), entry.getValue());
if (matches(metaEntry.getTagKey(), metaEntry.getTagValue(), tagPatterns)) {
if (resultField.size() < msg.getLimit()) {
Result r = new Result();
r.putTag(metaEntry.getTagKey(), metaEntry.getTagValue());
resultField.add(r);
final Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
try {
List<Result> resultField = new ArrayList<>();
Key start = new Key(Meta.VALUE_PREFIX + msg.getQuery());
Key end = start.followingKey(PartialKey.ROW);
Range range = new Range(start, end);
scanner.setRange(range);
tags.keySet().forEach(k -> scanner.fetchColumnFamily(new Text(k)));
int total = 0;
for (Entry<Key, Value> entry : scanner) {
Meta metaEntry = Meta.parse(entry.getKey(), entry.getValue());
if (matches(metaEntry.getTagKey(), metaEntry.getTagValue(), tagPatterns)) {
if (resultField.size() < msg.getLimit()) {
Result r = new Result();
r.putTag(metaEntry.getTagKey(), metaEntry.getTagValue());
resultField.add(r);
}
total++;
}
total++;
}
result.setResults(resultField);
result.setTotalResults(total);
result.setTime((int) (System.currentTimeMillis() - startMillis));
} finally {
if (scanner != null) {
scanner.close();
}
}
result.setResults(resultField);
result.setTotalResults(total);
result.setTime((int) (System.currentTimeMillis() - startMillis));
} catch (Exception ex) {
LOG.error("Error during lookup: " + ex.getMessage(), ex);
throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
@@ -756,70 +811,77 @@ public class DataStoreImpl implements DataStore {
public Set<Tag> getColumnFamilies(String metric, Map<String, String> requestedTags) throws TableNotFoundException {
Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
LOG.trace("Looking for requested tags: {}", tags);
Scanner meta = connector.createScanner(metaTable, Authorizations.EMPTY);
Text start = new Text(Meta.VALUE_PREFIX + metric);
Text end = new Text(Meta.VALUE_PREFIX + metric + "\\x0000");
end.append(new byte[] { (byte) 0xff }, 0, 1);
meta.setRange(new Range(start, end));
// Only look for the meta entries that match our tags, if any
boolean onlyFirstRow = false;
Entry<String, String> first = null;
// Set the columns on the meta scanner based on the first tag
// in the set of tags passed in the query. If no tags are present
// then we are only going to return the first tag name present in the
// meta table.
Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
if (tagIter.hasNext()) {
first = tagIter.next();
if (isTagValueRegex(first.getValue())) {
meta.fetchColumnFamily(new Text(first.getKey()));
} else {
meta.fetchColumn(new Text(first.getKey()), new Text(first.getValue()));
}
} else {
// grab all of the values found for the first tag for the metric
onlyFirstRow = true;
}
final boolean ONLY_RETURN_FIRST_TAG = onlyFirstRow;
Iterator<Entry<Key, Value>> iter = meta.iterator();
Iterator<Pair<String, String>> knownKeyValues = new Iterator<Pair<String, String>>() {
Text firstTag = null;
Text tagName = null;
Text tagValue = null;
@Override
public boolean hasNext() {
if (iter.hasNext()) {
Entry<Key, Value> metaEntry = iter.next();
if (null == firstTag) {
firstTag = metaEntry.getKey().getColumnFamily();
}
tagName = metaEntry.getKey().getColumnFamily();
tagValue = metaEntry.getKey().getColumnQualifier();
LOG.trace("Found tag entry {}={}", tagName, tagValue);
if (ONLY_RETURN_FIRST_TAG && !tagName.equals(firstTag)) {
return false;
}
return true;
Scanner meta = null;
try {
Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
LOG.trace("Looking for requested tags: {}", tags);
meta = connector.createScanner(metaTable, Authorizations.EMPTY);
Text start = new Text(Meta.VALUE_PREFIX + metric);
Text end = new Text(Meta.VALUE_PREFIX + metric + "\\x0000");
end.append(new byte[] { (byte) 0xff }, 0, 1);
meta.setRange(new Range(start, end));
// Only look for the meta entries that match our tags, if any
boolean onlyFirstRow = false;
Entry<String, String> first = null;
// Set the columns on the meta scanner based on the first tag
// in the set of tags passed in the query. If no tags are present
// then we are only going to return the first tag name present in the
// meta table.
Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
if (tagIter.hasNext()) {
first = tagIter.next();
if (isTagValueRegex(first.getValue())) {
meta.fetchColumnFamily(new Text(first.getKey()));
} else {
meta.fetchColumn(new Text(first.getKey()), new Text(first.getValue()));
}
return false;
} else {
// grab all of the values found for the first tag for the metric
onlyFirstRow = true;
}
final boolean ONLY_RETURN_FIRST_TAG = onlyFirstRow;
Iterator<Entry<Key, Value>> iter = meta.iterator();
Iterator<Pair<String, String>> knownKeyValues = new Iterator<Pair<String, String>>() {
@Override
public Pair<String, String> next() {
LOG.trace("Returning tag {}={}", tagName, tagValue);
return new Pair<>(tagName.toString(), tagValue.toString());
Text firstTag = null;
Text tagName = null;
Text tagValue = null;
@Override
public boolean hasNext() {
if (iter.hasNext()) {
Entry<Key, Value> metaEntry = iter.next();
if (null == firstTag) {
firstTag = metaEntry.getKey().getColumnFamily();
}
tagName = metaEntry.getKey().getColumnFamily();
tagValue = metaEntry.getKey().getColumnQualifier();
LOG.trace("Found tag entry {}={}", tagName, tagValue);
if (ONLY_RETURN_FIRST_TAG && !tagName.equals(firstTag)) {
return false;
}
return true;
}
return false;
}
@Override
public Pair<String, String> next() {
LOG.trace("Returning tag {}={}", tagName, tagValue);
return new Pair<>(tagName.toString(), tagValue.toString());
}
};
// Expand the list of tags in the meta table for this metric that
// matches
// the pattern of the first tag in the query. The resulting set of tags
// will be used to fetch specific columns from the metric table.
return expandTagValues(first, knownKeyValues);
} finally {
if (meta != null) {
meta.close();
}
};
// Expand the list of tags in the meta table for this metric that
// matches
// the pattern of the first tag in the query. The resulting set of tags
// will be used to fetch specific columns from the metric table.
return expandTagValues(first, knownKeyValues);
}
}
private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags, Set<Tag> colFamValues)
@@ -0,0 +1,116 @@
package timely.store;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.OptionDescriber;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.WrappingIterator;
import org.apache.commons.collections4.trie.PatriciaTrie;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.api.model.Meta;
public class MetaAgeOffIterator extends WrappingIterator implements OptionDescriber {
private static final Logger LOG = LoggerFactory.getLogger(MetaAgeOffIterator.class);
public static final String AGE_OFF_PREFIX = "ageoff.";
public static final String DEFAULT_AGEOFF_KEY = "default";
/* set in init */
private PatriciaTrie<Long> ageoffs = null;
private Long currentTime = 0L;
private Long defaultAgeOff = 0L;
private Long minAgeOff = Long.MAX_VALUE;
private Long maxAgeOff = Long.MIN_VALUE;
@Override
public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) {
MetaAgeOffIterator iter = new MetaAgeOffIterator();
iter.ageoffs = this.ageoffs;
iter.defaultAgeOff = this.defaultAgeOff;
iter.currentTime = this.currentTime;
iter.minAgeOff = this.minAgeOff;
iter.maxAgeOff = this.maxAgeOff;
return iter;
}
@Override
public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, IteratorEnvironment env)
throws IOException {
super.init(source, options, env);
validateOptions(options);
ageoffs = new PatriciaTrie<>();
options.forEach((k, v) -> {
if (k.startsWith(AGE_OFF_PREFIX)) {
String name = k.substring(AGE_OFF_PREFIX.length());
LOG.trace("Adding {} to Trie with value {}", name, Long.parseLong(v));
long ageoff = Long.parseLong(v);
this.minAgeOff = Math.min(this.minAgeOff, ageoff);
this.maxAgeOff = Math.max(this.maxAgeOff, ageoff);
ageoffs.put(name, ageoff);
}
});
defaultAgeOff = ageoffs.get(DEFAULT_AGEOFF_KEY);
currentTime = System.currentTimeMillis();
}
@Override
public void next() throws IOException {
super.next();
skipExpiredData();
}
@Override
public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) throws IOException {
super.seek(range, columnFamilies, inclusive);
skipExpiredData();
}
private void skipExpiredData() throws IOException {
boolean done = false;
while (!done && super.hasTop()) {
Key top = super.getTopKey();
long dataAge = this.currentTime - top.getTimestamp();
// If less than any configured ageoff, then we want to return this K,V
if (dataAge < this.minAgeOff) {
done = true;
} else {
String metricName = Meta.parse(top).getMetric();
long metaAgeOff = ageoffs.getOrDefault(metricName, defaultAgeOff);
if (dataAge > metaAgeOff) {
// data beyond ageOff, so skip
super.next();
} else {
done = true;
}
}
}
}
@Override
public IteratorOptions describeOptions() {
return new IteratorOptions("meta-age-off", "Iterator that ages off meta entries for Timely metrics",
Collections.singletonMap(AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY, "default age off days"),
Collections.singletonList("Additional meta age off properties where the value is specified as "
+ AGE_OFF_PREFIX
+ " the metric name and the value is an integer representing the number of days to keep"));
}
@Override
public boolean validateOptions(Map<String, String> options) {
if (null == options.get(AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY)) {
throw new IllegalArgumentException(DEFAULT_AGEOFF_KEY + " must be configured for MetaAgeOffFilter");
}
return true;
}
}
@@ -1,27 +1,123 @@
package timely.store;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.commons.configuration.BaseConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.api.model.Meta;
import timely.configuration.Accumulo;
import timely.configuration.Configuration;
public class MetaCacheImpl implements MetaCache {
private static Logger LOG = LoggerFactory.getLogger(MetaCacheImpl.class);
private static final Object DUMMY = new Object();
private volatile boolean closed = false;
private Cache<Meta, Object> cache = null;
private Configuration configuration;
private Timer timer = new Timer("MetaCacheImpl", true);
@Override
public void init(Configuration config) {
long expirationMinutes = config.getMetaCache().getExpirationMinutes();
int initialCapacity = config.getMetaCache().getInitialCapacity();
long maxCapacity = config.getMetaCache().getMaxCapacity();
cache = Caffeine.newBuilder().expireAfterAccess(expirationMinutes, TimeUnit.MINUTES)
.initialCapacity(initialCapacity).maximumSize(maxCapacity).build();
configuration = config;
cache = Caffeine.newBuilder().initialCapacity(1000).build();
long cacheRefreshMinutes = configuration.getMetaCache().getCacheRefreshMinutes();
if (cacheRefreshMinutes > 0) {
timer.schedule(new TimerTask() {
@Override
public void run() {
refreshCache(config.getMetaCache().getExpirationMinutes());
}
}, 60000, cacheRefreshMinutes * 60 * 1000);
}
}
private void refreshCache(long expirationMinutes) {
long oldestTimestamp = System.currentTimeMillis() - (expirationMinutes * 60 * 1000);
final BaseConfiguration apacheConf = new BaseConfiguration();
Accumulo accumuloConf = configuration.getAccumulo();
apacheConf.setProperty("instance.name", accumuloConf.getInstanceName());
apacheConf.setProperty("instance.zookeeper.host", accumuloConf.getZookeepers());
final ClientConfiguration aConf = new ClientConfiguration(Collections.singletonList(apacheConf));
String metaTable = configuration.getMetaTable();
Map<String, Map<String, Long>> metricMap = new HashMap<>();
Scanner scanner = null;
try {
final Instance instance = new ZooKeeperInstance(aConf);
Connector connector = instance.getConnector(accumuloConf.getUsername(),
new PasswordToken(accumuloConf.getPassword()));
scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
LOG.debug("Begin scanning " + metaTable);
Range metricRange = new Range(new Key(Meta.METRIC_PREFIX), true, new Key(Meta.METRIC_PREFIX + '\0'), false);
scanner.setRange(metricRange);
Set<String> allMetrics = new TreeSet<>();
for (Map.Entry<Key, Value> entry : scanner) {
Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.METRIC_PREFIX);
allMetrics.add(meta.getMetric());
}
for (String currMetric : allMetrics) {
Key begin = new Key(Meta.VALUE_PREFIX + currMetric);
Key end = new Key(Meta.VALUE_PREFIX + currMetric + '\0');
Range range = new Range(begin, true, end, false);
scanner.setRange(range);
Iterator<Map.Entry<Key, Value>> itr = scanner.iterator();
boolean maxedOutValues = false;
while (itr.hasNext() && !maxedOutValues) {
Map.Entry<Key, Value> entry = itr.next();
Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.VALUE_PREFIX);
String metric = meta.getMetric();
String tagKey = meta.getTagKey();
Map<String, Long> tagMap = metricMap.get(metric);
if (tagMap == null) {
tagMap = new HashMap<>();
metricMap.put(metric, tagMap);
}
long numTagValues = tagMap.getOrDefault(tagKey, 0l);
if (entry.getKey().getTimestamp() > oldestTimestamp) {
tagMap.put(tagKey, ++numTagValues);
if (numTagValues <= configuration.getMetaCache().getMaxTagValues()) {
cache.put(meta, DUMMY);
} else {
// found maxTagValues on this refresh, so remove this key
cache.invalidate(meta);
maxedOutValues = true;
}
} else {
// this entry is more than expirationMinutes old, so remove it
cache.invalidate(meta);
}
}
}
LOG.debug("Finished scanning " + metaTable);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
} finally {
if (scanner != null) {
scanner.close();
}
}
}
@Override
@@ -53,5 +149,4 @@ public class MetaCacheImpl implements MetaCache {
public boolean isClosed() {
return closed;
}
}
@@ -14,6 +14,7 @@ import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.commons.configuration.BaseConfiguration;
import org.springframework.boot.Banner.Mode;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import timely.api.model.Meta;
@@ -26,7 +27,7 @@ public class GetMetricTableSplitPoints {
public static void main(String[] args) throws Exception {
try (ConfigurableApplicationContext ctx = new SpringApplicationBuilder(SpringBootstrap.class)
.bannerMode(Mode.OFF).web(false).run(args)) {
.bannerMode(Mode.OFF).web(WebApplicationType.NONE).run(args)) {
Configuration conf = ctx.getBean(Configuration.class);
final BaseConfiguration apacheConf = new BaseConfiguration();
@@ -4,16 +4,22 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.test.util.EnvironmentTestUtils;
import org.springframework.boot.test.util.TestPropertyValues;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import timely.configuration.Configuration;
import timely.configuration.SpringBootstrap;
public class ConfigurationTest {
private final AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
private AnnotationConfigApplicationContext context;
@Before
public void createContext() {
context = new AnnotationConfigApplicationContext();
}
@After
public void closeContext() {
@@ -24,7 +30,7 @@ public class ConfigurationTest {
public void testMinimalConfiguration() throws Exception {
context.register(SpringBootstrap.class);
// @formatter:off
EnvironmentTestUtils.addEnvironment(this.context,
TestPropertyValues.of(
"timely.server.ip:127.0.0.1",
"timely.server.tcp-port:54321",
"timely.server.udp-port:54325",
@@ -38,7 +44,7 @@ public class ConfigurationTest {
"timely.accumulo.password:secret",
"timely.http.host:localhost",
"timely.security.serverSsl.use-generated-keypair:true",
"timely.metric-age-off-days[default]:7");
"timely.metric-age-off-days[default]:7").applyTo(this.context);
// @formatter:on
context.refresh();
Configuration config = this.context.getBean(Configuration.class);
@@ -62,7 +68,7 @@ public class ConfigurationTest {
public void testMissingSSLProperty() throws Exception {
context.register(SpringBootstrap.class);
// @formatter:off
EnvironmentTestUtils.addEnvironment(this.context,
TestPropertyValues.of(
"timely.server.ip:127.0.0.1",
"timely.server.tcp-port:54321",
"timely.server.udp-port:54325",
@@ -74,7 +80,7 @@ public class ConfigurationTest {
"timely.accumulo.instance-name:test",
"timely.accumulo.username:root",
"timely.accumulo.password:secret",
"timely.http.host:localhost");
"timely.http.host:localhost").applyTo(this.context);
// @formatter:on
context.refresh();
}
@@ -83,7 +89,7 @@ public class ConfigurationTest {
public void testSSLProperty() throws Exception {
context.register(SpringBootstrap.class);
// @formatter:off
EnvironmentTestUtils.addEnvironment(this.context,
TestPropertyValues.of(
"timely.server.ip:127.0.0.1",
"timely.server.tcp-port:54321",
"timely.server.udp-port:54325",
@@ -97,7 +103,7 @@ public class ConfigurationTest {
"timely.accumulo.password:secret",
"timely.http.host:localhost",
"timely.security.serverSsl.certificate-file:/tmp/foo",
"timely.security.serverSsl.key-file:/tmp/bar");
"timely.security.serverSsl.key-file:/tmp/bar").applyTo(this.context);
// @formatter:on
context.refresh();
}
@@ -30,19 +30,19 @@ public class AuthCacheTest {
@AfterClass
public static void after() {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Before
public void setup() throws Exception {
AuthCache.setSessionMaxAge(config.getSecurity());
AuthCache.configure(config.getSecurity());
UsernamePasswordAuthenticationToken token = new UsernamePasswordAuthenticationToken("test", "test1");
AuthenticationService.authenticate(token, cookie);
}
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test(expected = IllegalArgumentException.class)
@@ -22,8 +22,8 @@ public class AuthenticationServiceTest {
public static void before() {
config = TestConfiguration.createMinimalConfigurationForTest();
config.getSecurity().setSessionMaxAge(Integer.MAX_VALUE);
AuthCache.resetSessionMaxAge();
AuthCache.setSessionMaxAge(config.getSecurity());
AuthCache.resetConfiguration();
AuthCache.configure(config.getSecurity());
}
@Before
@@ -34,7 +34,7 @@ public class AuthenticationServiceTest {
@AfterClass
public static void after() {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test(expected = BadCredentialsException.class)
@@ -70,14 +70,14 @@ public class HttpRequestDecoderTest {
anonConfig = TestConfiguration.createMinimalConfigurationForTest();
anonConfig.getSecurity().setAllowAnonymousHttpAccess(true);
cookie = URLEncoder.encode(UUID.randomUUID().toString(), StandardCharsets.UTF_8.name());
AuthCache.setSessionMaxAge(config.getSecurity());
AuthCache.configure(config.getSecurity());
UsernamePasswordAuthenticationToken token = new UsernamePasswordAuthenticationToken("test", "test1");
AuthenticationService.authenticate(token, cookie);
}
@AfterClass
public static void after() {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Before
@@ -47,12 +47,12 @@ public class BasicAuthLoginRequestHandlerTest {
@BeforeClass
public static void before() throws Exception {
AuthCache.setSessionMaxAge(TestConfiguration.createMinimalConfigurationForTest().getSecurity());
AuthCache.configure(TestConfiguration.createMinimalConfigurationForTest().getSecurity());
}
@AfterClass
public static void after() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Before
@@ -64,14 +64,14 @@ public class WebSocketRequestDecoderTest {
anonConfig = TestConfiguration.createMinimalConfigurationForTest();
anonConfig.getSecurity().setAllowAnonymousWsAccess(true);
cookie = URLEncoder.encode(UUID.randomUUID().toString(), StandardCharsets.UTF_8.name());
AuthCache.setSessionMaxAge(config.getSecurity());
AuthCache.configure(config.getSecurity());
UsernamePasswordAuthenticationToken token = new UsernamePasswordAuthenticationToken("test", "test1");
AuthCache.put(cookie, new TimelyPrincipal(TimelyUser.ANONYMOUS_USER));
}
@AfterClass
public static void after() {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Before
@@ -0,0 +1,241 @@
package timely.store;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.SortedMapIterator;
import org.junit.Assert;
import org.junit.Test;
import timely.adapter.accumulo.MetaAdapter;
import timely.adapter.accumulo.MetricAdapter;
public class MetaAgeOffIteratorTest {
private static final Long TEST_TIME = System.currentTimeMillis() - 1000;
private static final Integer ONE_DAY = 86400000;
private static final Value EMPTY_VALUE = new Value(new byte[0]);
private static final Collection<ByteSequence> columnFamilies = new ArrayList<>();
@Test(expected = IllegalArgumentException.class)
public void testDefaultMissing() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
iter.init(source, options, null);
}
@Test
public void testDefault() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
seen++;
iter.next();
}
Assert.assertEquals(6, seen);
}
@Test
public void testMixed() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 5), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
seen++;
iter.next();
}
Assert.assertEquals(12, seen);
}
@Test
public void testAgeoffMixedMetricKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(
k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
seen++;
iter.next();
}
Assert.assertEquals(7, seen);
}
@Test
public void testAgeoffMixedTagKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag1", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag2", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag3", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag5", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag6", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag1", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag2", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag3", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag5", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag6", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(
k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
seen++;
iter.next();
}
Assert.assertEquals(7, seen);
}
@Test
public void testAgeoffMixedValueKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag1", "value1", TEST_TIME - (3 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag2", "value2", TEST_TIME - (2 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag3", "value3", TEST_TIME - (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag4", "value4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag5", "value5", TEST_TIME + (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag6", "value6", TEST_TIME + (2 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag1", "value1", TEST_TIME - (3 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag2", "value2", TEST_TIME - (2 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag3", "value3", TEST_TIME - (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag4", "value4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag5", "value5", TEST_TIME + (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag6", "value6", TEST_TIME + (2 * ONE_DAY)),
new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(
k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
seen++;
iter.next();
}
Assert.assertEquals(7, seen);
}
@Test
public void testSeekPastEndKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1));
iter.init(source, options, null);
iter.seek(new Range(new Key("sys.cpu.user"), true,
new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME + 3), new byte[0], new byte[0],
new byte[0], TEST_TIME + 3),
true), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
seen++;
iter.next();
}
Assert.assertEquals(0, seen);
}
}
@@ -41,6 +41,11 @@ public class CaptureChannelHandlerContext implements ChannelHandlerContext {
return null;
}
@Override
public <T> boolean hasAttr(AttributeKey<T> attributeKey) {
return false;
}
@Override
public Channel channel() {
return null;
@@ -65,7 +65,7 @@ public class OneWaySSLBase extends QueryBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
}
@@ -87,7 +87,7 @@ public class OneWaySSLBasicAuthAccessIT extends OneWaySSLBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -156,7 +156,7 @@ public class TwoWaySSLFailureIT extends QueryBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test(expected = UnauthorizedUserException.class)
@@ -160,7 +160,7 @@ public class TwoWaySSLIT extends QueryBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -135,7 +135,7 @@ public class TwoWaySSLOpenSSLIT extends QueryBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -43,7 +43,7 @@ public class TcpClientIT extends OneWaySSLBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -57,7 +57,7 @@ public class UdpClientIT extends MacITBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -73,7 +73,7 @@ public class WebSocketClientIT extends OneWaySSLBase {
@After
public void tearDown() throws Exception {
s.shutdown();
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
public void testWorkflow(WebSocketSubscriptionClient client) throws Exception {
@@ -8,6 +8,7 @@ import static org.junit.Assert.assertTrue;
import java.io.OutputStream;
import java.net.URL;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -20,6 +21,13 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.net.ssl.HttpsURLConnection;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import org.apache.commons.collections.CollectionUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.select.Elements;
@@ -130,12 +138,44 @@ public class HttpApiIT extends OneWaySSLBase {
String metrics = "https://localhost:54322/api/metrics";
// Test prefix matching
String result = query(metrics, "application/json");
assertEquals(expected, result);
Gson gson = new Gson();
JsonObject response = gson.fromJson(result, JsonObject.class);
JsonArray metricsArray = response.getAsJsonArray("metrics");
assertEquals(3, metricsArray.size());
for (JsonElement e : metricsArray) {
JsonObject metricObject = e.getAsJsonObject();
String metricName = metricObject.get("metric").getAsString();
Iterator<JsonElement> tagItr = metricObject.get("tags").getAsJsonArray().iterator();
Multimap<String, String> tagMap = HashMultimap.create();
while (tagItr.hasNext()) {
JsonObject tagObject = tagItr.next().getAsJsonObject();
tagMap.put(tagObject.get("key").getAsString(), tagObject.get("value").getAsString());
}
switch (metricName) {
case "sys.cpu.user":
assertEquals(2, tagMap.size());
assertTrue(CollectionUtils.isEqualCollection(Arrays.asList("value1"), tagMap.get("tag1")));
assertTrue(CollectionUtils.isEqualCollection(Arrays.asList("value2"), tagMap.get("tag2")));
break;
case "sys.cpu.idle":
assertEquals(2, tagMap.size());
assertTrue(CollectionUtils.isEqualCollection(Arrays.asList("value3"), tagMap.get("tag3")));
assertTrue(CollectionUtils.isEqualCollection(Arrays.asList("value4"), tagMap.get("tag4")));
break;
case "zzzz":
assertEquals(1, tagMap.size());
assertTrue(CollectionUtils.isEqualCollection(Arrays.asList("localhost"), tagMap.get("host")));
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
s.shutdown();
}
// @formatter:on
}
@Test
@@ -50,7 +50,7 @@ public class TimelyTcpIT extends MacITBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -64,7 +64,7 @@ public class TimelyUdpIT extends MacITBase {
@After
public void tearDown() throws Exception {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
@Test
@@ -176,7 +176,7 @@ public class WebSocketIT extends OneWaySSLBase {
@AfterClass
public static void after() {
AuthCache.resetSessionMaxAge();
AuthCache.resetConfiguration();
}
private EventLoopGroup group = null;
+4 -1
View File
@@ -46,9 +46,12 @@
<user-service id="usernamePasswordService">
<user name="test" password="test1" authorities="A,B,C" />
</user-service>
<beans:bean id="noOpPasswordEncoder" class="org.springframework.security.crypto.password.NoOpPasswordEncoder"/>
<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
<beans:property name="userDetailsService" ref="usernamePasswordService"/>
<beans:property name="passwordEncoder" ref="noOpPasswordEncoder"/>
</beans:bean>
<authentication-manager alias="authenticationManager">