Compare commits

..
48 changed files with 857 additions and 1702 deletions

No files matched your search

-4
View File
@@ -108,10 +108,6 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
</dependency>
<dependency>
<groupId>org.glassfish.tyrus</groupId>
<artifactId>tyrus-client</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
@@ -77,7 +77,7 @@ 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.TcpHealthChecker;
import timely.balancer.healthcheck.HealthChecker;
import timely.balancer.netty.http.HttpRelayHandler;
import timely.balancer.netty.tcp.TcpRelayHandler;
import timely.balancer.netty.udp.UdpRelayHandler;
@@ -257,7 +257,7 @@ public class Balancer {
if (this.metricResolver != null) {
try {
LOG.info("Closing metricResolver");
this.metricResolver.stop();
this.metricResolver.close();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
@@ -417,9 +417,8 @@ public class Balancer {
for (int x = 0; x < balancerConfig.getServer().getNumTcpPools(); x++) {
tcpClientPools.add(new TcpClientPool(this.balancerConfig));
}
TcpHealthChecker healthChecker = new TcpHealthChecker(this.balancerConfig, tcpClientPools.get(0));
HealthChecker healthChecker = new HealthChecker(this.balancerConfig, tcpClientPools.get(0));
this.metricResolver = new BalancedMetricResolver(curatorFramework, this.balancerConfig, healthChecker);
this.metricResolver.start();
final ServerBootstrap tcpServer = new ServerBootstrap();
tcpServer.group(tcpBossGroup, tcpWorkerGroup);
@@ -49,16 +49,6 @@ public class BalancerConfiguration {
private int serverSuccessesBeforeUp = 3;
private String metricAssignmentPersisterType = "HDFS";
private long balanceDelay = 900000;
private long balancePeriod = 900000;
private long persistDelay = 10000;
private long persistPeriod = 60000;
public ZooKeeper getZooKeeper() {
return zooKeeper;
}
@@ -130,44 +120,4 @@ public class BalancerConfiguration {
public void setServerSuccessesBeforeUp(int serverSuccessesBeforeUp) {
this.serverSuccessesBeforeUp = serverSuccessesBeforeUp;
}
public String getMetricAssignmentPersisterType() {
return metricAssignmentPersisterType;
}
public void setMetricAssignmentPersisterType(String metricAssignmentPersisterType) {
this.metricAssignmentPersisterType = metricAssignmentPersisterType;
}
public void setBalanceDelay(long balanceDelay) {
this.balanceDelay = balanceDelay;
}
public long getBalanceDelay() {
return balanceDelay;
}
public void setBalancePeriod(long balancePeriod) {
this.balancePeriod = balancePeriod;
}
public long getBalancePeriod() {
return balancePeriod;
}
public void setPersistDelay(long persistDelay) {
this.persistDelay = persistDelay;
}
public long getPersistDelay() {
return persistDelay;
}
public void setPersistPeriod(long persistPeriod) {
this.persistPeriod = persistPeriod;
}
public long getPersistPeriod() {
return persistPeriod;
}
}
@@ -42,10 +42,6 @@ public class TimelyBalancedHost {
return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort, arrivalRate);
}
static public TimelyBalancedHost of(String host, int basePort) {
return new TimelyBalancedHost(host, basePort, basePort + 1, basePort + 2, basePort + 3, null);
}
public void setBalancerConfig(BalancerConfiguration balancerConfig) {
this.balancerConfig = balancerConfig;
}
@@ -184,11 +180,4 @@ public class TimelyBalancedHost {
builder.append(isUp);
return builder.toString();
}
public static String toStringShort(TimelyBalancedHost tbh) {
if (tbh == null) {
return null;
}
return tbh.host + ":" + tbh.tcpPort;
}
}
@@ -1,12 +1,61 @@
package timely.balancer.healthcheck;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.connection.TimelyBalancedHost;
import timely.balancer.connection.tcp.TcpClientPool;
public interface HealthChecker {
public class HealthChecker {
public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts);
private static final Logger LOG = LoggerFactory.getLogger(HealthChecker.class);
private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
final private TCPServerHealthCheck tcpServerHealthCheck;
public void close() throws Exception;
public HealthChecker(BalancerConfiguration config, TcpClientPool tcpClientPool) {
this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
this.executorService.scheduleAtFixedRate(() -> {
synchronized (timelyHosts) {
for (TimelyBalancedHost h : timelyHosts) {
try {
if (tcpServerHealthCheck.isServerHealthy(h)) {
h.reportSuccess();
} else {
h.reportFailure();
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
h.reportFailure();
}
}
}
}, 0, config.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
}
public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
synchronized (timelyHosts) {
this.timelyHosts.clear();
this.timelyHosts.addAll(timelyHosts);
}
}
public void close() throws Exception {
this.executorService.shutdown();
try {
this.executorService.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
} finally {
if (!this.executorService.isTerminated()) {
this.executorService.shutdownNow();
}
}
}
}
@@ -1,62 +0,0 @@
package timely.balancer.healthcheck;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.connection.TimelyBalancedHost;
import timely.balancer.connection.tcp.TcpClientPool;
public class TcpHealthChecker implements HealthChecker {
private static final Logger LOG = LoggerFactory.getLogger(TcpHealthChecker.class);
private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
final private TCPServerHealthCheck tcpServerHealthCheck;
public TcpHealthChecker(BalancerConfiguration config, TcpClientPool tcpClientPool) {
this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
this.executorService.scheduleAtFixedRate(() -> {
synchronized (timelyHosts) {
for (TimelyBalancedHost h : timelyHosts) {
try {
if (tcpServerHealthCheck.isServerHealthy(h)) {
h.reportSuccess();
} else {
h.reportFailure();
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
h.reportFailure();
}
}
}
}, 0, config.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
}
@Override
public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
synchronized (timelyHosts) {
this.timelyHosts.clear();
this.timelyHosts.addAll(timelyHosts);
}
}
public void close() throws Exception {
this.executorService.shutdown();
try {
this.executorService.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
} finally {
if (!this.executorService.isTerminated()) {
this.executorService.shutdownNow();
}
}
}
}
@@ -4,14 +4,11 @@ import static timely.Server.SERVICE_DISCOVERY_PATH;
import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
import static timely.balancer.Balancer.ASSIGNMENTS_LOCK_PATH;
import static timely.balancer.Balancer.LEADER_LATCH_PATH;
import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.ASSIGN_FALLBACK_SEQUENTIAL;
import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.ASSIGN_ROUND_ROBIN;
import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.HOST_DOWN_ROUND_ROBIN;
import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.HOST_REMOVED;
import static timely.balancer.resolver.eventing.MetricAssignedEvent.Reason.REBALANCE;
import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
import java.io.FileNotFoundException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -32,9 +29,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import com.csvreader.CsvReader;
import com.csvreader.CsvWriter;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
@@ -49,6 +49,11 @@ import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.ServiceCacheListener;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,23 +62,16 @@ import timely.balancer.ArrivalRate;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.connection.TimelyBalancedHost;
import timely.balancer.healthcheck.HealthChecker;
import timely.balancer.resolver.eventing.MetricAssignedCallback;
import timely.balancer.resolver.eventing.MetricAssignedEvent;
import timely.balancer.resolver.eventing.MetricBalanceCallback;
import timely.balancer.resolver.eventing.MetricBalanceEvent;
import timely.balancer.resolver.eventing.MetricHostCallback;
import timely.balancer.resolver.eventing.MetricHostEvent;
public class BalancedMetricResolver implements MetricResolver {
private static final Logger LOG = LoggerFactory.getLogger(BalancedMetricResolver.class);
private Map<String, TimelyBalancedHost> metricToHostMap = new TreeMap<>();
private Set<String> unassignedMetrics = new HashSet<>();
private Map<String, ArrivalRate> metricMap = new HashMap<>();
private ReentrantReadWriteLock balancerLock = new ReentrantReadWriteLock();
private List<TimelyBalancedHost> serverList = new ArrayList<>();
private Random r = new Random();
private final HealthChecker healthChecker;
final private HealthChecker healthChecker;
private ScheduledExecutorService assignmentExecutor = Executors.newScheduledThreadPool(2);
private ScheduledExecutorService arrivalRateExecutor = Executors.newScheduledThreadPool(2);
private int roundRobinCounter = 0;
@@ -81,46 +79,38 @@ public class BalancedMetricResolver implements MetricResolver {
private DistributedAtomicValue nonCachedMetricsIP;
private ReentrantReadWriteLock nonCachedMetricsLocalLock = new ReentrantReadWriteLock();
private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
private BalancerConfiguration balancerConfig;
private LeaderLatch leaderLatch;
private AtomicBoolean isLeader = new AtomicBoolean(false);
private InterProcessReadWriteLock assignmentsIPRWLock;
private MetricAssignmentPerister assignmentPerister;
private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
final private BalancerConfiguration balancerConfig;
final private CuratorFramework curatorFramework;
private List<MetricAssignedCallback> metricAssignedCallbacks = new ArrayList<>();
private List<MetricBalanceCallback> metricBalanceCallbacks = new ArrayList<>();
private List<MetricHostCallback> metricHostCallbacks = new ArrayList<>();
public enum BalanceType {
HIGH_LOW, HIGH_AVG, AVG_LOW;
private enum BalanceType {
HIGH_LOW, HIGH_AVG, AVG_LOW
}
public BalancedMetricResolver(CuratorFramework curatorFramework, BalancerConfiguration balancerConfig,
HealthChecker healthChecker) {
this.curatorFramework = curatorFramework;
this.balancerConfig = balancerConfig;
this.healthChecker = healthChecker;
}
public void start() {
assignmentsIPRWLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
testIPRWLock(curatorFramework, assignmentsIPRWLock, ASSIGNMENTS_LOCK_PATH);
this.assignmentPerister = new MetricAssignmentPerister(this, balancerConfig, metricToHostMap, nonCachedMetrics,
curatorFramework, assignmentsIPRWLock, nonCachedMetricsLocalLock, assignmentsLastUpdatedLocal,
balancerLock);
startLeaderLatch(curatorFramework);
startServiceListener(curatorFramework);
assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH,
new RetryForever(1000));
TreeCacheListener assignmentListener = (curatorFramework1, event) -> {
LOG.debug("Handling assignments event {}. assignmentsLastUpdated:{}", event.getType().toString(),
new Date(assignmentPerister.getAssignmentsLastUpdatedByLeader()));
if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
assignmentPerister.readAssignments(true);
TreeCacheListener assignmentListener = new TreeCacheListener() {
@Override
public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
LOG.debug("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
readAssignmentsFromHdfs(true);
}
}
};
@@ -160,7 +150,7 @@ public class BalancedMetricResolver implements MetricResolver {
}
readNonCachedMetricsIP();
assignmentPerister.readAssignments(false);
readAssignmentsFromHdfs(false);
assignmentExecutor.scheduleAtFixedRate(() -> {
if (isLeader.get()) {
@@ -170,28 +160,28 @@ public class BalancedMetricResolver implements MetricResolver {
LOG.error(e.getMessage(), e);
}
}
}, balancerConfig.getBalanceDelay(), balancerConfig.getBalancePeriod(), TimeUnit.MILLISECONDS);
}, 15, 15, TimeUnit.MINUTES);
assignmentExecutor.scheduleAtFixedRate(() -> {
try {
if (isLeader.get()) {
long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
long lastUpdateByLeader = assignmentPerister.getAssignmentsLastUpdatedByLeader();
if (lastLocalUpdate > lastUpdateByLeader) {
LOG.debug("Leader writing assignments lastLocalUpdate ({}) > lastUpdateByLeader ({})",
new Date(lastLocalUpdate), new Date(lastUpdateByLeader));
assignmentPerister.writeAssignments();
long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
if (lastLocalUpdate > lastHdfsUpdate) {
LOG.debug("Leader writing assignments to hdfs lastLocalUpdate ({}) > lastHdfsUpdate ({})",
new Date(lastLocalUpdate), new Date(lastHdfsUpdate));
writeAssignmentsToHdfs();
} else {
LOG.trace("Leader not writing assignments lastLocalUpdate ({}) <= lastUpdateByLeader ({})",
new Date(lastLocalUpdate), new Date(lastUpdateByLeader));
LOG.trace("Leader not writing assignments to hdfs lastLocalUpdate ({}) <= lastHdfsUpdate ({})",
new Date(lastLocalUpdate), new Date(lastHdfsUpdate));
}
} else {
assignmentPerister.readAssignments(true);
readAssignmentsFromHdfs(true);
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}, balancerConfig.getPersistDelay(), balancerConfig.getPersistPeriod(), TimeUnit.MILLISECONDS);
}, 10, 60, TimeUnit.SECONDS);
}
private void readNonCachedMetricsIP() {
@@ -249,7 +239,7 @@ public class BalancedMetricResolver implements MetricResolver {
public void isLeader() {
LOG.info("this balancer is the leader");
isLeader.set(true);
assignmentPerister.writeAssignments();
writeAssignmentsToHdfs();
}
@Override
@@ -276,10 +266,10 @@ public class BalancedMetricResolver implements MetricResolver {
ServerDetails pl = si.getPayload();
TimelyBalancedHost tbh = TimelyBalancedHost.of(pl.getHost(), pl.getTcpPort(), pl.getHttpPort(),
pl.getWsPort(), pl.getUdpPort(), new ArrivalRate(arrivalRateExecutor));
LOG.info("adding service {} host:{}", si.getId(), TimelyBalancedHost.toStringShort(tbh));
LOG.info("adding service {} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", si.getId(),
pl.getHost(), pl.getTcpPort(), pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort());
tbh.setBalancerConfig(balancerConfig);
serverList.add(tbh);
metricHostEvent(tbh, MetricHostEvent.ActionType.ADDED);
}
healthChecker.setTimelyHosts(serverList);
} finally {
@@ -315,8 +305,9 @@ public class BalancedMetricResolver implements MetricResolver {
availableHosts.remove(h);
} else {
itr.remove();
metricHostEvent(h, MetricHostEvent.ActionType.REMOVED);
LOG.info("removing service {}", TimelyBalancedHost.toStringShort(h));
LOG.info("removing service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}",
h.getHost(), h.getTcpPort(), h.getHost(), h.getTcpPort(), h.getHttpPort(),
h.getWsPort(), h.getUdpPort());
for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
if (e.getValue().equals(h)) {
reassignMetrics.add(e.getKey());
@@ -328,22 +319,19 @@ public class BalancedMetricResolver implements MetricResolver {
// add new hosts that were not previously known
for (TimelyBalancedHost h : availableHosts) {
LOG.info("adding service {}", TimelyBalancedHost.toStringShort(h));
LOG.info("adding service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}",
h.getHost(), h.getTcpPort(), h.getHost(), h.getTcpPort(), h.getHttpPort(),
h.getWsPort(), h.getUdpPort());
serverList.add(h);
metricHostEvent(h, MetricHostEvent.ActionType.ADDED);
rebalanceNeeded = true;
}
healthChecker.setTimelyHosts(serverList);
if (!availableHosts.isEmpty()) {
reassignMetrics.addAll(unassignedMetrics);
unassignedMetrics.clear();
}
if (isLeader.get()) {
for (String s : reassignMetrics) {
TimelyBalancedHost h = getRoundRobinHost(null);
assignMetric(s, h, HOST_REMOVED);
LOG.debug("Assigned server removed. Assigning {} to server {}", s,
TimelyBalancedHost.toStringShort(h));
assignMetric(s, h);
LOG.debug("Assigned server removed. Assigning {} to server {}:{}", s, h.getTcpPort());
}
}
@@ -354,7 +342,7 @@ public class BalancedMetricResolver implements MetricResolver {
if (rebalanceNeeded) {
balance();
}
assignmentPerister.writeAssignments();
writeAssignmentsToHdfs();
}
}
@@ -414,7 +402,7 @@ public class BalancedMetricResolver implements MetricResolver {
return tbh;
}
protected TimelyBalancedHost getRoundRobinHost(TimelyBalancedHost notThisOne) {
private TimelyBalancedHost getRoundRobinHost(TimelyBalancedHost notThisOne) {
TimelyBalancedHost tbh = null;
balancerLock.readLock().lock();
try {
@@ -536,10 +524,8 @@ public class BalancedMetricResolver implements MetricResolver {
int numReassigned = 0;
if (isLeader.get()) {
metricBalanceEvent(MetricBalanceEvent.ProgressType.BEGIN, balanceType, 0);
balancerLock.writeLock().lock();
try {
metricBalanceEvent(MetricBalanceEvent.ProgressType.BALANCER_LOCK_ACQUIRED, balanceType, 0);
Map<String, ArrivalRate> tempMetricMap = new HashMap<>();
tempMetricMap.putAll(metricMap);
@@ -569,8 +555,8 @@ public class BalancedMetricResolver implements MetricResolver {
}
}
LOG.trace("focusHost {} desiredChange:{} rateSortedMetrics.size():{}",
TimelyBalancedHost.toStringShort(h), desiredChange, rateSortedMetrics.size());
LOG.trace("focusHost {}:{} desiredChange:{} rateSortedMetrics.size():{}", h.getHost(),
h.getTcpPort(), desiredChange, rateSortedMetrics.size());
boolean doneWithHost = false;
while (!doneWithHost) {
@@ -599,7 +585,7 @@ public class BalancedMetricResolver implements MetricResolver {
} else {
String metric = e.getValue();
Double metricRate = e.getKey();
assignMetric(metric, candidateHost, REBALANCE);
assignMetric(metric, candidateHost);
numReassigned++;
calculatedRates.put(candidateHost, calculatedRates.get(candidateHost) + metricRate);
calculatedRates.put(h, calculatedRates.get(h) - metricRate);
@@ -608,9 +594,9 @@ public class BalancedMetricResolver implements MetricResolver {
itr.remove();
tempMetricMap.remove(metric);
LOG.info(
"rebalancing: reassigning metric:{} rate:{} from server {} to {} remaining delta {}",
metric, metricRate, TimelyBalancedHost.toStringShort(h),
TimelyBalancedHost.toStringShort(candidateHost), desiredChange);
"rebalancing: reassigning metric:{} rate:{} from server {}:{} to {}:{} remaining delta {}",
metric, metricRate, h.getHost(), h.getTcpPort(), candidateHost.getHost(),
candidateHost.getTcpPort(), desiredChange);
}
}
}
@@ -633,7 +619,6 @@ public class BalancedMetricResolver implements MetricResolver {
}
} finally {
balancerLock.writeLock().unlock();
metricBalanceEvent(MetricBalanceEvent.ProgressType.END, balanceType, numReassigned);
}
}
return numReassigned;
@@ -706,7 +691,7 @@ public class BalancedMetricResolver implements MetricResolver {
}
}
protected boolean shouldCache(String metricName) {
private boolean shouldCache(String metricName) {
if (StringUtils.isBlank(metricName)) {
return false;
@@ -768,7 +753,7 @@ public class BalancedMetricResolver implements MetricResolver {
}
if (StringUtils.isBlank(metric)) {
tbh = getRoundRobinHost(null);
tbh = getRandomHost(null);
} else {
tbh = metricToHostMap.get(metric);
if (tbh == null) {
@@ -777,17 +762,17 @@ public class BalancedMetricResolver implements MetricResolver {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
}
assignMetric(metric, tbh, ASSIGN_ROUND_ROBIN);
assignMetric(metric, tbh);
} else if (!tbh.isUp()) {
TimelyBalancedHost oldTbh = tbh;
tbh = getRoundRobinHost(oldTbh);
LOG.debug("rebalancing from host that is down: reassigning metric {} from server {} to {}", metric,
TimelyBalancedHost.toStringShort(oldTbh), TimelyBalancedHost.toStringShort(tbh));
tbh = getLeastUsedHost();
LOG.debug("rebalancing from host that is down: reassigning metric {} from server {}:{} to {}:{}",
metric, oldTbh.getHost(), oldTbh.getTcpPort(), tbh.getHost(), tbh.getTcpPort());
if (!balancerLock.isWriteLockedByCurrentThread()) {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
}
assignMetric(metric, tbh, HOST_DOWN_ROUND_ROBIN);
assignMetric(metric, tbh);
}
}
@@ -804,7 +789,7 @@ public class BalancedMetricResolver implements MetricResolver {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
}
assignMetric(metric, tbh, ASSIGN_FALLBACK_SEQUENTIAL);
assignMetric(metric, tbh);
}
}
} finally {
@@ -833,23 +818,20 @@ public class BalancedMetricResolver implements MetricResolver {
metric = null;
}
if (tbh == null) {
tbh = getRoundRobinHost(null);
if (!balancerLock.isWriteLockedByCurrentThread()) {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
if (tbh == null || !tbh.isUp()) {
for (int x = 0; tbh == null && x < serverList.size(); x++) {
tbh = serverList.get(Math.abs(r.nextInt() & Integer.MAX_VALUE) % serverList.size());
if (!tbh.isUp()) {
tbh = null;
}
}
assignMetric(metric, tbh, ASSIGN_ROUND_ROBIN);
} else if (!tbh.isUp()) {
TimelyBalancedHost oldTbh = tbh;
tbh = getRoundRobinHost(oldTbh);
LOG.debug("rebalancing from host that is down: reassigning metric {} from server {} to {}", metric,
TimelyBalancedHost.toStringShort(oldTbh), TimelyBalancedHost.toStringShort(tbh));
if (!balancerLock.isWriteLockedByCurrentThread()) {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
if (tbh != null && StringUtils.isNotBlank(metric)) {
if (!balancerLock.isWriteLockedByCurrentThread()) {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
}
assignMetric(metric, tbh);
}
assignMetric(metric, tbh, HOST_DOWN_ROUND_ROBIN);
}
// if all else fails
@@ -865,7 +847,7 @@ public class BalancedMetricResolver implements MetricResolver {
balancerLock.readLock().unlock();
balancerLock.writeLock().lock();
}
assignMetric(metric, tbh, ASSIGN_FALLBACK_SEQUENTIAL);
assignMetric(metric, tbh);
}
}
} finally {
@@ -878,7 +860,7 @@ public class BalancedMetricResolver implements MetricResolver {
return tbh;
}
protected TimelyBalancedHost findHost(String host, int tcpPort) {
private TimelyBalancedHost findHost(String host, int tcpPort) {
TimelyBalancedHost tbh = null;
for (TimelyBalancedHost h : serverList) {
if (h.getHost().equals(host) && h.getTcpPort() == tcpPort) {
@@ -889,73 +871,199 @@ public class BalancedMetricResolver implements MetricResolver {
return tbh;
}
private void assignMetric(String metric, TimelyBalancedHost tbh, MetricAssignedEvent.Reason reason) {
private void createAssignmentFile() {
try {
LOG.info("Creating assignment file: " + balancerConfig.getAssignmentFile());
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
fs.create(assignmentFile);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}
private void readAssignmentsFromHdfs(boolean checkIfNecessary) {
try {
long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
if (checkIfNecessary) {
if (lastHdfsUpdate <= lastLocalUpdate) {
LOG.debug("Not reading assignments from hdfs lastHdfsUpdate ({}) <= lastLocalUpdate ({})",
new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
return;
}
}
LOG.debug("Reading assignments from hdfs lastHdfsUpdate ({}) > lastLocalUpdate ({})",
new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
// proceed with reading from HDFS
Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
try {
boolean acquired = false;
while (!acquired) {
acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
}
balancerLock.writeLock().lock();
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
FSDataInputStream iStream = fs.open(assignmentFile);
CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
reader.setUseTextQualifier(false);
// skip the headers
boolean success = true;
success = reader.readHeaders();
while (success) {
success = reader.readRecord();
String[] nextLine = reader.getValues();
if (nextLine.length >= 3) {
String metric = nextLine[0];
String host = nextLine[1];
int tcpPort = Integer.parseInt(nextLine[2]);
TimelyBalancedHost tbh = findHost(host, tcpPort);
if (tbh == null) {
tbh = getRoundRobinHost(null);
} else {
LOG.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
}
assignedMetricToHostMap.put(metric, tbh);
}
}
metricToHostMap.clear();
for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
if (StringUtils.isNotBlank(e.getKey()) && e.getValue() != null) {
if (shouldCache(e.getKey())) {
metricToHostMap.put(e.getKey(), e.getValue());
}
} else {
Exception e1 = new IllegalStateException(
"Bad assignment metric:" + e.getKey() + " host:" + e.getValue());
LOG.warn(e1.getMessage(), e1);
}
}
nonCachedMetricsLocalLock.readLock().lock();
try {
Iterator<Map.Entry<String, TimelyBalancedHost>> itr = metricToHostMap.entrySet().iterator();
while (itr.hasNext()) {
Map.Entry<String, TimelyBalancedHost> e = itr.next();
if (nonCachedMetrics.contains(e.getKey())) {
itr.remove();
}
}
} finally {
nonCachedMetricsLocalLock.readLock().unlock();
}
assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
LOG.info("Read {} assignments from hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
new Date(assignmentsLastUpdatedLocal.get()));
} catch (FileNotFoundException e) {
createAssignmentFile();
metricToHostMap.clear();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
} finally {
balancerLock.writeLock().unlock();
try {
assignmentsIPRWLock.readLock().release();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}
}
private void writeAssignmentsToHdfs() {
CsvWriter writer = null;
try {
boolean acquired = false;
while (!acquired) {
acquired = assignmentsIPRWLock.writeLock().acquire(60, TimeUnit.SECONDS);
}
balancerLock.readLock().lock();
nonCachedMetricsLocalLock.readLock().lock();
try {
if (!metricToHostMap.isEmpty()) {
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
if (!fs.exists(assignmentFile.getParent())) {
fs.mkdirs(assignmentFile.getParent());
}
FSDataOutputStream oStream = fs.create(assignmentFile, true);
writer = new CsvWriter(oStream, ',', Charset.forName("UTF-8"));
writer.setUseTextQualifier(false);
writer.write("metric");
writer.write("host");
writer.write("tcpPort");
writer.endRecord();
for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
if (!nonCachedMetrics.contains(e.getKey())) {
writer.write(e.getKey());
writer.write(e.getValue().getHost());
writer.write(Integer.toString(e.getValue().getTcpPort()));
writer.endRecord();
LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
e.getValue().getTcpPort());
}
}
long now = System.currentTimeMillis();
assignmentsLastUpdatedLocal.set(now);
assignmentsLastUpdatedInHdfs.trySet(now);
if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
assignmentsLastUpdatedInHdfs.forceSet(now);
}
LOG.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})",
metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
}
} finally {
nonCachedMetricsLocalLock.readLock().unlock();
balancerLock.readLock().unlock();
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
} finally {
if (writer != null) {
writer.close();
}
try {
assignmentsIPRWLock.writeLock().release();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}
}
private void assignMetric(String metric, TimelyBalancedHost tbh) {
if (isLeader.get()) {
if (StringUtils.isNotBlank(metric) && tbh != null) {
balancerLock.writeLock().lock();
try {
metricAssignedEvent(metric, metricToHostMap.get(metric), tbh, reason);
metricToHostMap.put(metric, tbh);
assignmentsLastUpdatedLocal.set(System.currentTimeMillis());
} finally {
balancerLock.writeLock().unlock();
}
} else {
unassignedMetrics.add(metric);
// Exception e = new IllegalStateException("Bad assignment metric:" + metric + "
// host:" + tbh);
LOG.warn("Bad assignment metric:" + metric + " host:" + tbh);
Exception e = new IllegalStateException("Bad assignment metric:" + metric + " host:" + tbh);
LOG.warn(e.getMessage(), e);
}
}
}
public void registerCallback(MetricAssignedCallback callback) {
this.metricAssignedCallbacks.add(callback);
}
public void registerCallback(MetricBalanceCallback callback) {
this.metricBalanceCallbacks.add(callback);
}
public void registerCallback(MetricHostCallback callback) {
this.metricHostCallbacks.add(callback);
}
protected void metricAssignedEvent(String metric, TimelyBalancedHost losing, TimelyBalancedHost gaining,
MetricAssignedEvent.Reason reason) {
if (!metricAssignedCallbacks.isEmpty()) {
MetricAssignedEvent event = new MetricAssignedEvent(metric, losing, gaining, reason);
for (MetricAssignedCallback callback : metricAssignedCallbacks) {
callback.onEvent(event);
}
}
}
protected void metricBalanceEvent(MetricBalanceEvent.ProgressType progressType,
BalancedMetricResolver.BalanceType balanceType, long numReassigned) {
if (!metricBalanceCallbacks.isEmpty()) {
MetricBalanceEvent event = new MetricBalanceEvent(progressType, balanceType, numReassigned);
for (MetricBalanceCallback callback : metricBalanceCallbacks) {
callback.onEvent(event);
}
}
}
protected void metricHostEvent(TimelyBalancedHost host, MetricHostEvent.ActionType actionType) {
if (!metricHostCallbacks.isEmpty()) {
MetricHostEvent event = new MetricHostEvent(host, actionType);
for (MetricHostCallback callback : metricHostCallbacks) {
callback.onEvent(event);
}
}
}
protected void writeAssignments(Map<String, TimelyBalancedHost> metricToHostMap) {
assignmentPerister.writeAssignments(metricToHostMap);
}
@Override
public void stop() {
public void close() throws Exception {
this.assignmentExecutor.shutdown();
try {
@@ -1,320 +0,0 @@
package timely.balancer.resolver;
import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import com.csvreader.CsvReader;
import com.csvreader.CsvWriter;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.retry.RetryForever;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.connection.TimelyBalancedHost;
import timely.balancer.resolver.eventing.MetricAssignedEvent;
public class MetricAssignmentPerister {
private final Logger LOG = LoggerFactory.getLogger(getClass());
private enum PeristerType {
HDFS, FILE
}
// passed in through constructor
BalancedMetricResolver balancedMetricResolver;
private BalancerConfiguration balancerConfig;
private Map<String, TimelyBalancedHost> metricToHostMap;
private Set<String> nonCachedMetrics;
private InterProcessReadWriteLock assignmentsIPRWLock;
private ReentrantReadWriteLock nonCachedMetricsLocalLock;
private ReentrantReadWriteLock balancerLock;
private AtomicLong assignmentsLastUpdatedLocal;
// local
private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
public MetricAssignmentPerister(BalancedMetricResolver balancedMetricResolver, BalancerConfiguration balancerConfig,
Map<String, TimelyBalancedHost> metricToHostMap, Set<String> nonCachedMetrics,
CuratorFramework curatorFramework, InterProcessReadWriteLock assignmentsIPRWLock,
ReentrantReadWriteLock nonCachedMetricsLocalLock, AtomicLong assignmentsLastUpdatedLocal,
ReentrantReadWriteLock balancerLock) {
this.balancedMetricResolver = balancedMetricResolver;
this.balancerConfig = balancerConfig;
this.metricToHostMap = metricToHostMap;
this.nonCachedMetrics = nonCachedMetrics;
this.assignmentsIPRWLock = assignmentsIPRWLock;
this.nonCachedMetricsLocalLock = nonCachedMetricsLocalLock;
this.assignmentsLastUpdatedLocal = assignmentsLastUpdatedLocal;
this.balancerLock = balancerLock;
assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH,
new RetryForever(1000));
}
public long getAssignmentsLastUpdatedByLeader() throws Exception {
return assignmentsLastUpdatedInHdfs.get().postValue();
}
public void readAssignments(boolean checkIfNecessary) {
try {
long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
if (checkIfNecessary) {
if (lastHdfsUpdate <= lastLocalUpdate) {
LOG.debug("Not reading assignments from storage lastStorageUpdate ({}) <= lastLocalUpdate ({})",
new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
return;
}
}
LOG.debug("Reading assignments from storage lastStorageUpdate ({}) > lastLocalUpdate ({})",
new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
// proceed with reading from storage
try (InputStream inputStream = getInputStream(balancerConfig)) {
boolean acquired = false;
while (!acquired) {
acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
}
balancerLock.writeLock().lock();
// get potential new list from implementation
Map<String, TimelyBalancedHost> assignedMetricToHostMap = readFromStream(inputStream);
metricToHostMap.clear();
for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
if (StringUtils.isNotBlank(e.getKey()) && e.getValue() != null) {
if (balancedMetricResolver.shouldCache(e.getKey())) {
balancedMetricResolver.metricAssignedEvent(e.getKey(), null, e.getValue(),
MetricAssignedEvent.Reason.ASSIGN_FILE);
metricToHostMap.put(e.getKey(), e.getValue());
}
} else {
Exception e1 = new IllegalStateException(
"Bad assignment metric:" + e.getKey() + " host:" + e.getValue());
LOG.warn(e1.getMessage(), e1);
}
}
nonCachedMetricsLocalLock.readLock().lock();
try {
Iterator<Map.Entry<String, TimelyBalancedHost>> itr = metricToHostMap.entrySet().iterator();
while (itr.hasNext()) {
Map.Entry<String, TimelyBalancedHost> e = itr.next();
if (nonCachedMetrics.contains(e.getKey())) {
itr.remove();
}
}
} finally {
nonCachedMetricsLocalLock.readLock().unlock();
}
assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
LOG.info("Read {} assignments from storage lastStorageUpdate = lastLocalUpdate ({})",
metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
} catch (FileNotFoundException e) {
createAssignmentFile();
metricToHostMap.clear();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
} finally {
balancerLock.writeLock().unlock();
try {
assignmentsIPRWLock.readLock().release();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}
}
public void writeAssignments() {
writeAssignments(metricToHostMap);
}
protected void writeAssignments(Map<String, TimelyBalancedHost> metricToHostMap) {
try {
boolean acquired = false;
while (!acquired) {
acquired = assignmentsIPRWLock.writeLock().acquire(60, TimeUnit.SECONDS);
}
balancerLock.readLock().lock();
nonCachedMetricsLocalLock.readLock().lock();
try (OutputStream outputStream = getOutputStream(balancerConfig)) {
if (!metricToHostMap.isEmpty()) {
writeToStream(outputStream);
long now = System.currentTimeMillis();
assignmentsLastUpdatedLocal.set(now);
assignmentsLastUpdatedInHdfs.trySet(now);
if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
assignmentsLastUpdatedInHdfs.forceSet(now);
}
LOG.debug("Wrote {} assignments to storage lastStorageUpdate = lastLocalUpdate ({})",
metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
}
} finally {
nonCachedMetricsLocalLock.readLock().unlock();
balancerLock.readLock().unlock();
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
} finally {
try {
assignmentsIPRWLock.writeLock().release();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}
}
public Map<String, TimelyBalancedHost> readFromStream(InputStream inputStream) throws IOException {
Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
try {
CsvReader reader = new CsvReader(inputStream, ',', Charset.forName("UTF-8"));
reader.setUseTextQualifier(false);
// skip the headers
boolean success = true;
success = reader.readHeaders();
while (success) {
success = reader.readRecord();
String[] nextLine = reader.getValues();
if (nextLine.length >= 3) {
String metric = nextLine[0];
String host = nextLine[1];
int tcpPort = Integer.parseInt(nextLine[2]);
TimelyBalancedHost tbh = balancedMetricResolver.findHost(host, tcpPort);
if (tbh == null) {
tbh = balancedMetricResolver.getRoundRobinHost(null);
} else {
LOG.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
}
assignedMetricToHostMap.put(metric, tbh);
}
}
} catch (IOException e) {
LOG.error(e.getMessage(), e);
throw e;
}
return assignedMetricToHostMap;
}
private void writeToStream(OutputStream outputStream) throws IOException {
CsvWriter writer = null;
try {
writer = new CsvWriter(outputStream, ',', Charset.forName("UTF-8"));
writer.setUseTextQualifier(false);
writer.write("metric");
writer.write("host");
writer.write("tcpPort");
writer.endRecord();
for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
if (!nonCachedMetrics.contains(e.getKey())) {
writer.write(e.getKey());
writer.write(e.getValue().getHost());
writer.write(Integer.toString(e.getValue().getTcpPort()));
writer.endRecord();
LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
e.getValue().getTcpPort());
}
}
} finally {
if (writer != null) {
writer.close();
}
}
}
public void createAssignmentFile() {
LOG.info("Creating assignment file: " + balancerConfig.getAssignmentFile());
PeristerType peristerType = PeristerType.valueOf(balancerConfig.getMetricAssignmentPersisterType());
switch (peristerType) {
case HDFS:
try {
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
fs.create(assignmentFile);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
break;
case FILE:
try {
File assignmentFile = new File(balancerConfig.getAssignmentFile());
String parentPath = assignmentFile.getParent();
File directory = new File(parentPath);
if (!directory.exists()) {
directory.mkdirs();
}
if (!assignmentFile.exists()) {
assignmentFile.createNewFile();
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
break;
}
}
public InputStream getInputStream(BalancerConfiguration balancerConfiguration) throws Exception {
PeristerType peristerType = PeristerType.valueOf(balancerConfiguration.getMetricAssignmentPersisterType());
switch (peristerType) {
case HDFS:
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
Path assignmentFile = new Path(balancerConfiguration.getAssignmentFile());
return fs.open(assignmentFile);
case FILE:
return new FileInputStream(new File(balancerConfiguration.getAssignmentFile()));
default:
throw new IllegalArgumentException(
"Unknown metricPersisterType " + balancerConfiguration.getMetricAssignmentPersisterType());
}
}
public OutputStream getOutputStream(BalancerConfiguration balancerConfiguration) throws IOException {
PeristerType peristerType = PeristerType.valueOf(balancerConfiguration.getMetricAssignmentPersisterType());
switch (peristerType) {
case HDFS:
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
Path assignmentFile = new Path(balancerConfiguration.getAssignmentFile());
if (!fs.exists(assignmentFile.getParent())) {
fs.mkdirs(assignmentFile.getParent());
}
return fs.create(assignmentFile, true);
case FILE:
return new FileOutputStream(new File(balancerConfiguration.getAssignmentFile()));
default:
throw new IllegalArgumentException(
"Unknown metricPersisterType " + balancerConfiguration.getMetricAssignmentPersisterType());
}
}
}
@@ -8,8 +8,5 @@ public interface MetricResolver {
public TimelyBalancedHost getHostPortKey(String metric);
public void start();
public void stop();
public void close() throws Exception;
}
@@ -1,7 +0,0 @@
package timely.balancer.resolver.eventing;
public abstract class MetricAssignedCallback {
public abstract void onEvent(MetricAssignedEvent event);
}
@@ -1,39 +0,0 @@
package timely.balancer.resolver.eventing;
import timely.balancer.connection.TimelyBalancedHost;
public class MetricAssignedEvent {
public enum Reason {
ASSIGN_ROUND_ROBIN, HOST_DOWN_ROUND_ROBIN, ASSIGN_FALLBACK_SEQUENTIAL, HOST_REMOVED, REBALANCE, ASSIGN_FILE
}
private final String metric;
private final TimelyBalancedHost losingHost;
private final TimelyBalancedHost gainingHost;
private final Reason reason;
public MetricAssignedEvent(String metric, TimelyBalancedHost losingHost, TimelyBalancedHost gainingHost,
Reason reason) {
this.metric = metric;
this.losingHost = losingHost;
this.gainingHost = gainingHost;
this.reason = reason;
}
public String getMetric() {
return metric;
}
public TimelyBalancedHost getLosingHost() {
return losingHost;
}
public TimelyBalancedHost getGainingHost() {
return gainingHost;
}
public Reason getReason() {
return reason;
}
}
@@ -1,7 +0,0 @@
package timely.balancer.resolver.eventing;
public abstract class MetricBalanceCallback {
public abstract void onEvent(MetricBalanceEvent event);
}
@@ -1,33 +0,0 @@
package timely.balancer.resolver.eventing;
import timely.balancer.resolver.BalancedMetricResolver;
public class MetricBalanceEvent {
public enum ProgressType {
BEGIN, BALANCER_LOCK_ACQUIRED, END
}
private final ProgressType progressType;
private final BalancedMetricResolver.BalanceType balanceType;
private final Long numRessigned;
public MetricBalanceEvent(ProgressType progressType, BalancedMetricResolver.BalanceType balanceType,
long numReassigned) {
this.progressType = progressType;
this.balanceType = balanceType;
this.numRessigned = numReassigned;
}
public BalancedMetricResolver.BalanceType getBalanceType() {
return balanceType;
}
public ProgressType getProgressType() {
return progressType;
}
public Long getNumRessigned() {
return numRessigned;
}
}
@@ -1,7 +0,0 @@
package timely.balancer.resolver.eventing;
public abstract class MetricHostCallback {
public abstract void onEvent(MetricHostEvent event);
}
@@ -1,26 +0,0 @@
package timely.balancer.resolver.eventing;
import timely.balancer.connection.TimelyBalancedHost;
public class MetricHostEvent {
public enum ActionType {
ADDED, REMOVED, SUCCESS, FAILURE
}
private final TimelyBalancedHost timelyBalancedHost;
private final ActionType actionType;
public MetricHostEvent(TimelyBalancedHost timelyBalancedHost, ActionType actionType) {
this.timelyBalancedHost = timelyBalancedHost;
this.actionType = actionType;
}
public TimelyBalancedHost getTimelyBalancedHost() {
return timelyBalancedHost;
}
public ActionType getActionType() {
return actionType;
}
}
-5
View File
@@ -10,9 +10,4 @@
<Method name="getBuf" />
<Bug pattern="EI_EXPOSE_REP" />
</Match>
<Match>
<Class name="timely.balancer.resolver.MetricAssignmentPerister" />
<Method name="createAssignmentFile" />
<Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
</Match>
</FindBugsFilter>
@@ -1,76 +0,0 @@
package timely.balancer.resolver;
import static timely.Server.SERVICE_DISCOVERY_PATH;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.ServiceInstanceBuilder;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.ServerDetails;
import timely.balancer.connection.TimelyBalancedHost;
public class ServiceRegistry {
private static final Logger LOG = LoggerFactory.getLogger(ServiceRegistry.class);
public static void registerService(CuratorFramework curatorFramework, TimelyBalancedHost tbh) {
try {
try {
Stat stat = curatorFramework.checkExists().forPath(SERVICE_DISCOVERY_PATH);
if (stat == null) {
curatorFramework.create().creatingParentContainersIfNeeded().forPath(SERVICE_DISCOVERY_PATH);
}
} catch (Exception e) {
LOG.error(e.getMessage());
}
ServiceInstance<ServerDetails> serviceInstance = serviceInstance(tbh);
ServiceDiscovery<ServerDetails> discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class)
.client(curatorFramework).basePath(SERVICE_DISCOVERY_PATH).build();
discovery.start();
discovery.registerService(serviceInstance);
} catch (Exception e) {
LOG.error(e.getMessage());
}
}
public static void unregisterService(CuratorFramework curatorFramework, TimelyBalancedHost tbh) {
try {
try {
Stat stat = curatorFramework.checkExists().forPath(SERVICE_DISCOVERY_PATH);
if (stat == null) {
curatorFramework.create().creatingParentContainersIfNeeded().forPath(SERVICE_DISCOVERY_PATH);
}
} catch (Exception e) {
LOG.error(e.getMessage());
}
ServiceInstance<ServerDetails> serviceInstance = serviceInstance(tbh);
ServiceDiscovery<ServerDetails> discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class)
.client(curatorFramework).basePath(SERVICE_DISCOVERY_PATH).build();
discovery.start();
discovery.unregisterService(serviceInstance);
} catch (Exception e) {
LOG.error(e.getMessage());
}
}
private static ServiceInstance<ServerDetails> serviceInstance(TimelyBalancedHost config) throws Exception {
ServerDetails serverDetails = new ServerDetails();
serverDetails.setHost(config.getHost());
serverDetails.setTcpPort(config.getTcpPort());
serverDetails.setHttpPort(config.getHttpPort());
serverDetails.setWsPort(config.getWsPort());
serverDetails.setUdpPort(config.getUdpPort());
ServiceInstanceBuilder<ServerDetails> builder = ServiceInstance.builder();
String serviceName = config.getHost() + ":" + config.getTcpPort();
ServiceInstance<ServerDetails> serviceInstance = builder.id(serviceName).name("timely-server")
.address(config.getHost()).port(config.getTcpPort()).payload(serverDetails).build();
return serviceInstance;
}
}
@@ -1,270 +0,0 @@
package timely.balancer.resolver;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.server.ServerConfig;
import org.apache.zookeeper.server.ZooKeeperServerMain;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.SocketUtils;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.connection.TimelyBalancedHost;
import timely.balancer.resolver.eventing.MetricAssignedCallback;
import timely.balancer.resolver.eventing.MetricAssignedEvent;
import timely.balancer.resolver.eventing.MetricBalanceCallback;
import timely.balancer.resolver.eventing.MetricBalanceEvent;
import timely.balancer.resolver.eventing.MetricHostCallback;
import timely.balancer.resolver.eventing.MetricHostEvent;
public class TestBalancedMetricResolver {
private final Logger LOG = LoggerFactory.getLogger(getClass());
private CuratorFramework curatorFramework;
private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HHmmss.SSS");
private Random r = new Random(System.currentTimeMillis());
private List<TimelyBalancedHost> timelyHosts;
private Map<MetricAssignedEvent, Integer> metricBalanceEventCounter = new HashMap<>();
@Before
public void startup() {
QuorumPeerConfig currentConfig = createZookeeper();
InetSocketAddress address = currentConfig.getClientPortAddress();
String zkServer = address.getHostName() + ":" + address.getPort();
curatorFramework = CuratorFrameworkFactory.newClient(zkServer, 1000, 5000, new RetryNTimes(2, 50));
curatorFramework.start();
timelyHosts = new ArrayList<>();
}
@After
public void shutdown() {
curatorFramework.close();
}
private QuorumPeerConfig createZookeeper() {
Properties startupProperties = new Properties();
File file = new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID());
file.deleteOnExit();
startupProperties.setProperty("dataDir", file.getAbsolutePath());
startupProperties.setProperty("clientPort", String.valueOf(SocketUtils.findAvailableTcpPort(3000, 4000)));
QuorumPeerConfig quorumConfiguration = new QuorumPeerConfig();
try {
quorumConfiguration.parseProperties(startupProperties);
} catch (Exception e) {
throw new RuntimeException(e);
}
ZooKeeperServerMain zooKeeperServer = new ZooKeeperServerMain();
final ServerConfig configuration = new ServerConfig();
configuration.readFrom(quorumConfiguration);
new Thread() {
public void run() {
try {
zooKeeperServer.runFromConfig(configuration);
} catch (IOException e) {
LOG.error("ZooKeeper Failed", e);
}
}
}.start();
return quorumConfiguration;
}
private BalancerConfiguration getBalancerConfiguration() {
BalancerConfiguration balancerConfiguration = new BalancerConfiguration();
balancerConfiguration.setMetricAssignmentPersisterType("FILE");
File file = new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID());
file.deleteOnExit();
balancerConfiguration.setAssignmentFile(file.getAbsolutePath());
balancerConfiguration.setBalanceDelay(0);
balancerConfiguration.setBalancePeriod(1000);
balancerConfiguration.setPersistDelay(0);
balancerConfiguration.setPersistDelay(1000);
balancerConfiguration.setCheckServerHealthInterval(10);
return balancerConfiguration;
}
private void createTimelyHosts(BalancerConfiguration balancerConfiguration, int numHosts) {
for (int port = 1000, x = 0; x < numHosts; x++, port += 10) {
TimelyBalancedHost tbh = TimelyBalancedHost.of("localhost", port);
tbh.setBalancerConfig(balancerConfiguration);
timelyHosts.add(tbh);
ServiceRegistry.registerService(curatorFramework, tbh);
}
}
private void registerCallbacks(BalancedMetricResolver resolver) {
resolver.registerCallback(new MetricAssignedCallback() {
@Override
public void onEvent(MetricAssignedEvent event) {
LOG.debug(String.format("%s metric:%s lose:%s:%d gain:%s:%d reason:%s",
event.getClass().getSimpleName(), event.getMetric(),
event.getLosingHost() == null ? null : event.getLosingHost().getHost(),
event.getLosingHost() == null ? null : event.getLosingHost().getTcpPort(),
event.getGainingHost().getHost(), event.getGainingHost().getTcpPort(), event.getReason()));
}
});
resolver.registerCallback(new MetricHostCallback() {
@Override
public void onEvent(MetricHostEvent event) {
LOG.debug(String.format("%s host:%s:%d type:%s", event.getClass().getSimpleName(),
event.getTimelyBalancedHost().getHost(), event.getTimelyBalancedHost().getTcpPort(),
event.getActionType()));
}
});
resolver.registerCallback(new MetricBalanceCallback() {
@Override
public void onEvent(MetricBalanceEvent event) {
LOG.debug(String.format("%s type:%s progress:%s reassigned:%d", event.getClass().getSimpleName(),
event.getBalanceType(), event.getProgressType(), event.getNumRessigned()));
}
});
}
private TimelyBalancedHost randomServerDown(TestHealthChecker healthChecker, int numHosts) {
TimelyBalancedHost tbh = timelyHosts.get(r.nextInt(numHosts));
LOG.debug("Marking server down:" + TimelyBalancedHost.toStringShort(tbh));
healthChecker.serverDown(tbh);
return tbh;
}
private TimelyBalancedHost randomServerToggle(TestHealthChecker healthChecker, int numHosts) {
TimelyBalancedHost tbh = timelyHosts.get(r.nextInt(numHosts));
LOG.debug("Marking server " + (tbh.isUp() ? "down:" : "up:") + TimelyBalancedHost.toStringShort(tbh));
healthChecker.serverToggle(tbh);
return tbh;
}
private TimelyBalancedHost randomServerUnregister(int numHosts) {
TimelyBalancedHost tbh = timelyHosts.get(r.nextInt(numHosts));
LOG.debug("Unregistering:" + TimelyBalancedHost.toStringShort(tbh));
ServiceRegistry.unregisterService(curatorFramework, tbh);
return tbh;
}
@Test
public void test1() {
LOG.debug("Configure test");
int numHosts = 10;
BalancerConfiguration balancerConfiguration = getBalancerConfiguration();
createTimelyHosts(balancerConfiguration, numHosts);
TestHealthChecker healthChecker = new TestHealthChecker(balancerConfiguration, timelyHosts);
BalancedMetricResolver balancedMetricResolver = new BalancedMetricResolver(curatorFramework,
balancerConfiguration, healthChecker);
registerCallbacks(balancedMetricResolver);
LOG.debug("Start balancedMetricResolver");
balancedMetricResolver.start();
long now = System.currentTimeMillis();
int numMetrics = 100;
for (int x = 0; x < numMetrics; x++) {
balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", x));
}
LOG.debug("Begin monitoring");
long cycle = 0;
List<TimelyBalancedHost> serversUnregistered = new ArrayList<>();
while (System.currentTimeMillis() < now + 15000) {
try {
Thread.sleep(100);
for (int x = 0; x < 10000; x++) {
int mNum = r.nextInt(numMetrics);
balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", mNum));
}
// if (cycle % 100 == 0) {
// randomServerDown(healthChecker, numHosts);
// }
// if (cycle % 2 == 0) {
// randomServerToggle(healthChecker, numHosts);
// }
if (cycle % 100 == 0) {
serversUnregistered.add(randomServerUnregister(numHosts));
}
if (cycle % 500 == 0) {
serversUnregistered.stream().forEach(tbh -> ServiceRegistry.registerService(curatorFramework, tbh));
serversUnregistered.clear();
}
} catch (InterruptedException e) {
}
cycle++;
}
LOG.debug("Stop balancedMetricResolver");
balancedMetricResolver.stop();
}
@Test
public void test2() {
LOG.debug("Configure test");
int numHosts = 10;
BalancerConfiguration balancerConfiguration = getBalancerConfiguration();
createTimelyHosts(balancerConfiguration, numHosts);
TestHealthChecker healthChecker = new TestHealthChecker(balancerConfiguration, timelyHosts);
BalancedMetricResolver balancedMetricResolver = new BalancedMetricResolver(curatorFramework,
balancerConfiguration, healthChecker);
registerCallbacks(balancedMetricResolver);
LOG.debug("Start balancedMetricResolver");
balancedMetricResolver.start();
long now = System.currentTimeMillis();
int numMetrics = 100;
for (int x = 0; x < numMetrics; x++) {
balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", x));
}
LOG.debug("Begin monitoring");
long cycle = 0;
List<TimelyBalancedHost> serversUnregistered = new ArrayList<>();
while (System.currentTimeMillis() < now + 15000) {
try {
Thread.sleep(100);
for (int x = 0; x < 10000; x++) {
int mNum = r.nextInt(numMetrics);
balancedMetricResolver.getHostPortKeyIngest(String.format("metric%03d", mNum));
}
// if (cycle % 100 == 0) {
// randomServerDown(healthChecker, numHosts);
// }
// if (cycle % 2 == 0) {
// randomServerToggle(healthChecker, numHosts);
// }
if (cycle % 100 == 0) {
serversUnregistered.add(randomServerUnregister(numHosts));
}
if (cycle % 500 == 0) {
serversUnregistered.stream().forEach(tbh -> ServiceRegistry.registerService(curatorFramework, tbh));
serversUnregistered.clear();
}
} catch (InterruptedException e) {
}
cycle++;
}
LOG.debug("Stop balancedMetricResolver");
balancedMetricResolver.stop();
}
}
@@ -1,89 +0,0 @@
package timely.balancer.resolver;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.balancer.configuration.BalancerConfiguration;
import timely.balancer.connection.TimelyBalancedHost;
import timely.balancer.healthcheck.HealthChecker;
import timely.balancer.healthcheck.TcpHealthChecker;
public class TestHealthChecker implements HealthChecker {
private final Logger LOG = LoggerFactory.getLogger(TcpHealthChecker.class);
private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
private Map<TimelyBalancedHost, Boolean> serverUpMap = Collections.synchronizedMap(new HashMap<>());
private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
public TestHealthChecker(BalancerConfiguration config, List<TimelyBalancedHost> timelyHosts) {
setTimelyHosts(timelyHosts);
this.executorService.scheduleAtFixedRate(() -> {
synchronized (timelyHosts) {
for (TimelyBalancedHost h : timelyHosts) {
try {
if (TestHealthChecker.this.isServerHealthy(h)) {
h.reportSuccess();
} else {
h.reportFailure();
}
} catch (Exception e) {
LOG.error(e.getMessage(), e);
h.reportFailure();
}
}
}
}, 0, config.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
}
@Override
public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
synchronized (timelyHosts) {
this.timelyHosts.clear();
this.timelyHosts.addAll(timelyHosts);
for (TimelyBalancedHost h : timelyHosts) {
serverUpMap.put(h, true);
}
}
}
public boolean isServerHealthy(TimelyBalancedHost h) {
return serverUpMap.get(h);
}
public void serverUp(TimelyBalancedHost h) {
serverUpMap.put(h, true);
}
public void serverDown(TimelyBalancedHost h) {
serverUpMap.put(h, false);
}
public void serverToggle(TimelyBalancedHost h) {
synchronized (serverUpMap) {
serverUpMap.put(h, !serverUpMap.get(h));
}
}
@Override
public void close() throws Exception {
this.executorService.shutdown();
try {
this.executorService.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
} finally {
if (!this.executorService.isTerminated()) {
this.executorService.shutdownNow();
}
}
}
}
-18
View File
@@ -1,18 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="CONSOLE" target="SYSTEM_OUT">
<PatternLayout pattern="%d{yyyyMMdd HHmmss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
</Appenders>
<Loggers>
<Logger name="org.apache.accumulo.core.client.impl.ThriftTransportPool" level="OFF" additivity="false" />
<Logger name="org.apache.accumulo.fate.zookeeper.ZooCache" level="OFF" additivity="false" />
<Logger name="org.apache.zookeeper" level="OFF" additivity="false" />
<Logger name="timely.balancer.resolver" level="TRACE" additivity="true" />
<Logger name="timely.balancer.resolver.TestBalancedMetricResolver" level="DEBUG" additivity="true" />
<Root level="ERROR">
<AppenderRef ref="CONSOLE"/>
</Root>
</Loggers>
</Configuration>
@@ -22,7 +22,7 @@ import timely.model.parse.TagParser;
@JsonRootName("metric")
@JsonInclude(JsonInclude.Include.NON_NULL)
@JsonPropertyOrder({ "name" })
public class Metric implements Serializable {
public class Metric implements Serializable, ObjectSizeOf {
private static final long serialVersionUID = 1L;
@@ -186,4 +186,9 @@ public class Metric implements Serializable {
return this.metric;
}
}
@Override
public long sizeInBytes() {
return 0;
}
}
@@ -2,13 +2,16 @@ package timely.model;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Stack;
import java.util.logging.Logger;
import org.apache.commons.lang3.StringUtils;
/**
* A simple interface that objects can implement to return the object size.
*/
@@ -22,15 +25,29 @@ public interface ObjectSizeOf {
public static class ObjectInstance {
private Object o;
private String name = "";
public ObjectInstance(Object _o) {
this.o = _o;
}
public ObjectInstance(Object _o, String name) {
this.o = _o;
this.name = name;
}
public Object getObject() {
return o;
}
public void setName(String name) {
this.name = name;
}
public String getName() {
return name;
}
@Override
public int hashCode() {
return System.identityHashCode(o);
@@ -64,110 +81,134 @@ public interface ObjectSizeOf {
*
* @return an approximation of the object size
*/
public static long getObjectSize(Object o, boolean useSizeInBytesMethod, boolean debug) {
long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(),
useSizeInBytesMethod, debug);
return estSize;
}
public static long getObjectSize(Object o) {
long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(), true, false);
return estSize;
return getObjectSize(o, true, false);
}
public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack,
boolean useSizeInBytesMethod, boolean debug) {
long totalSize = 0;
stack.add(new ObjectInstance(o));
while (!stack.isEmpty()) {
long size = 0;
ObjectInstance oi = stack.pop();
o = oi.getObject();
if (o != null && !visited.contains(oi)) {
visited.add(oi);
try {
if (useSizeInBytesMethod) {
try {
if (o instanceof ObjectSizeOf) {
size = ((ObjectSizeOf) o).sizeInBytes();
} else {
Method sizeInBytes = o.getClass().getMethod("sizeInBytes", (Class<?>[]) null);
size = (Long) sizeInBytes.invoke(o);
}
} catch (Throwable t) {
// ok, lets do this the hard way...
}
}
if (size == 0) {
// the hard way...
// do not include Class related objects or
// reflection objects
if ((o instanceof Class) || (o instanceof ClassLoader) || (o.getClass().getPackage() != null
&& o.getClass().getPackage().getName().startsWith("java.lang.reflect"))) {
size = 0;
} else if (o instanceof Number || o instanceof Boolean || o instanceof Character) {
size = NUMBER_SIZE;
} else {
// lets do a simple sizing
Class<?> c = o.getClass();
if (c.isArray()) {
size = ARRAY_OVERHEAD;
int length = Array.getLength(o);
if (c.getComponentType().isPrimitive()) {
size += length * getPrimitiveObjectSize(c.getComponentType());
} else {
size += length * REFERENCE;
for (int i = 0; i < length; i++) {
Object element = Array.get(o, i);
if (element != null) {
stack.add(new ObjectInstance(element));
}
}
}
} else {
size += OBJECT_OVERHEAD;
while (c != null) {
for (Field field : c.getDeclaredFields()) {
if (Modifier.isStatic(field.getModifiers())) {
continue;
}
if (field.getType().isPrimitive()) {
size += getPrimitiveObjectSize(field.getType());
} else {
size += REFERENCE;
boolean accessible = field.isAccessible();
field.setAccessible(true);
try {
Object fieldObject = field.get(o);
if (fieldObject != null) {
stack.push(new ObjectInstance(fieldObject));
}
} catch (Exception e) {
// cannot get to field, so
// ignore it in this size
// calculation
e.printStackTrace();
}
field.setAccessible(accessible);
}
}
c = c.getSuperclass();
}
}
size = roundUp(size);
}
}
} catch (Throwable t) {
log.warning("Unable to determine object size for " + o);
}
if (debug) {
System.out.println(o.getClass().getCanonicalName() + "=" + size);
}
}
totalSize += size;
public static long getObjectSize(Object o, boolean useSizeInBytesMethod, boolean debug) {
List<String> output = null;
if (debug) {
output = new ArrayList<>();
}
long size = getObjectSize(new ObjectInstance(o, o.getClass().getSimpleName()), new HashSet<>(), 0,
useSizeInBytesMethod, output);
if (debug) {
Collections.reverse(output);
for (String s : output)
System.out.println(s);
}
return size;
}
return totalSize;
public static long getObjectSize(ObjectInstance oi, Set<ObjectInstance> visited, int indent,
boolean useSizeInBytesMethod, List<String> output) {
long size = 0;
Object o = oi.getObject();
int arrayOverhead = 0;
int objectOverhead = 0;
int reference = 0;
boolean roundUp = false;
if (o != null && !visited.contains(oi)) {
visited.add(oi);
try {
if (useSizeInBytesMethod) {
if (o instanceof ObjectSizeOf) {
size = ((ObjectSizeOf) o).sizeInBytes();
if (output != null && size > 0) {
output.add(StringUtils.repeat(' ', indent) + oi.getName() + " [" + size + "] ("
+ o.getClass().getCanonicalName() + ") [ObjectSizeOf]");
}
}
}
if (size == 0) {
// the hard way...
// do not include Class related objects or
// reflection objects
if ((o instanceof Class) || (o instanceof ClassLoader) || (o.getClass().getPackage() != null
&& o.getClass().getPackage().getName().startsWith("java.lang.reflect"))) {
size = 0;
} else if (o instanceof Number || o instanceof Boolean || o instanceof Character) {
size = NUMBER_SIZE;
} else {
// lets do a simple sizing
Class<?> c = o.getClass();
if (c.isArray()) {
arrayOverhead++;
int length = Array.getLength(o);
if (c.getComponentType().isPrimitive()) {
size += length * getPrimitiveObjectSize(c.getComponentType());
} else {
reference += length;
for (int i = 0, index = length - 1; i < length; i++, index--) {
Object element = Array.get(o, index);
if (element != null) {
size += getObjectSize(
new ObjectInstance(element, oi.getName() + "[" + index + "]"),
visited, indent + 2, useSizeInBytesMethod, output);
}
}
}
} else {
objectOverhead++;
while (c != null) {
for (Field field : c.getDeclaredFields()) {
if (Modifier.isStatic(field.getModifiers())) {
continue;
}
if (field.getType().isPrimitive()) {
long primSize = getPrimitiveObjectSize(field.getType());
if (output != null) {
output.add(StringUtils.repeat(' ', indent + 2) + field.getName() + " ["
+ primSize + "] (" + field.getType().getCanonicalName() + ")");
}
size += primSize;
} else {
reference++;
boolean accessible = field.isAccessible();
field.setAccessible(true);
try {
Object fieldObject = field.get(o);
if (fieldObject != null) {
size += getObjectSize(
new ObjectInstance(fieldObject, field.getName()), visited,
indent + 2, useSizeInBytesMethod, output);
}
} catch (Exception e) {
// cannot get to field, so
// ignore it in this size
// calculation
e.printStackTrace();
}
field.setAccessible(accessible);
}
}
c = c.getSuperclass();
}
}
roundUp = true;
}
}
long underlyingObjects = size;
long overheads = REFERENCE * reference + OBJECT_OVERHEAD * objectOverhead
+ ARRAY_OVERHEAD * arrayOverhead;
long addedOverheads = size + overheads;
if (roundUp) {
size = roundUp(addedOverheads);
} else {
size = addedOverheads;
}
if (output != null) {
output.add(StringUtils.repeat(' ', indent) + oi.getName() + " [" + size + "] ("
+ o.getClass().getCanonicalName() + ") = " + underlyingObjects + "+" + overheads + "+"
+ (size - addedOverheads) + " REF=" + reference + " OBJ=" + objectOverhead + " ARRAY="
+ arrayOverhead + " ROUND=" + (size - addedOverheads));
}
} catch (Throwable t) {
log.warning("Unable to determine object size for " + o);
}
}
return size;
}
public static long roundUp(long size) {
@@ -191,6 +232,42 @@ public interface ObjectSizeOf {
return 0;
}
}
public static long getStringSize(String s) {
long size = 0;
size += ObjectSizeOf.Sizer.OBJECT_OVERHEAD;
size += ObjectSizeOf.Sizer.REFERENCE;
size += ObjectSizeOf.Sizer.getPrimitiveObjectSize(int.class);
try {
// change with Java 9
if (String.class.getDeclaredField("coder") != null) {
size += ObjectSizeOf.Sizer.getPrimitiveObjectSize(byte.class);
}
} catch (Exception e) {
}
size += roundUp(ARRAY_OVERHEAD + (s.length() * ObjectSizeOf.Sizer.getPrimitiveObjectSize(byte.class)));
return roundUp(size);
}
}
public static void main(String[] args) {
Metric m = new Metric("mymetric", 100000l, 32);
m.addTag(new Tag("key1", "value1"));
m.addTag(new Tag("key2", "value2"));
m.addTag(new Tag("key3", "value3"));
m.addTag(new Tag("key4", "value4"));
m.addTag(new Tag("key5", "value5"));
long newSize1 = ObjectSizeOf.Sizer.getObjectSize(m, false, true);
System.out.println("new size #1: " + newSize1);
System.out.println();
System.out.println();
System.out.println();
long newSize2 = ObjectSizeOf.Sizer.getObjectSize(m, true, true);
System.out.println("new size #2: " + newSize2);
}
}
+32 -1
View File
@@ -18,7 +18,7 @@ import timely.model.parse.TagParser;
* Tag consists of key value pair
*/
@JsonRootName("tag")
public class Tag implements Comparable<Tag>, Serializable {
public class Tag implements Comparable<Tag>, ObjectSizeOf, Serializable {
private static final long serialVersionUID = 1L;
@@ -27,6 +27,12 @@ public class Tag implements Comparable<Tag>, Serializable {
private String key;
private String value;
@JsonIgnore
private boolean mustRecalculateLength = true;
@JsonIgnore
private long length;
public Tag() {
}
@@ -55,6 +61,7 @@ public class Tag implements Comparable<Tag>, Serializable {
public void set(String key, String value) {
this.key = key;
this.value = value;
this.mustRecalculateLength = true;
}
@JsonIgnore
@@ -64,6 +71,7 @@ public class Tag implements Comparable<Tag>, Serializable {
public void setKey(String key) {
this.key = key;
this.mustRecalculateLength = true;
}
@JsonIgnore
@@ -73,6 +81,7 @@ public class Tag implements Comparable<Tag>, Serializable {
public void setValue(String value) {
this.value = value;
this.mustRecalculateLength = true;
}
/**
@@ -106,4 +115,26 @@ public class Tag implements Comparable<Tag>, Serializable {
public int compareTo(Tag other) {
return new CompareToBuilder().append(this.key, other.key).append(this.value, other.value).toComparison();
}
@JsonIgnore
@Override
public long sizeInBytes() {
if (this.mustRecalculateLength) {
calculateLength();
this.mustRecalculateLength = false;
}
return this.length;
}
private void calculateLength() {
long l = 0;
l += 2 * Sizer.REFERENCE;
l += Sizer.OBJECT_OVERHEAD;
l += Sizer.getPrimitiveObjectSize(long.class);
l += Sizer.getPrimitiveObjectSize(boolean.class);
l += Sizer.getStringSize(key);
l += Sizer.getStringSize(value);
this.length = Sizer.roundUp(l);
}
}
+13 -1
View File
@@ -5,13 +5,14 @@ import java.util.Comparator;
import java.util.concurrent.TimeUnit;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Objects;
/**
* Value consists of a long timestamp and a double measurement
*/
public class Value implements Comparator<Value>, Serializable {
public class Value implements Comparator<Value>, ObjectSizeOf, Serializable {
private static final long serialVersionUID = 1L;
@@ -90,4 +91,15 @@ public class Value implements Comparator<Value>, Serializable {
public String toString() {
return "Value{" + "timestamp=" + timestamp + ", measure=" + measure + "}";
}
@JsonIgnore
@Override
public long sizeInBytes() {
long l = 0;
l += Sizer.OBJECT_OVERHEAD;
l += 2 * Sizer.REFERENCE;
l += Sizer.NUMBER_SIZE;
l += Sizer.NUMBER_SIZE;
return Sizer.roundUp(l);
}
}
-4
View File
@@ -108,10 +108,6 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
</dependency>
<dependency>
<groupId>org.glassfish.tyrus</groupId>
<artifactId>tyrus-client</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
+11 -28
View File
@@ -132,6 +132,7 @@ public class Server {
private static final String OS_VERSION = "os.version";
private static final String WS_PATH = "/websocket";
private SslContext sslCtx = null;
private CuratorFramework curatorFramework;
protected static final CountDownLatch LATCH = new CountDownLatch(1);
static ConfigurableApplicationContext applicationContext;
@@ -238,7 +239,7 @@ public class Server {
Server server = new Server(conf);
try {
server.run();
server.run(createSSLContext(conf));
LATCH.await();
} catch (final InterruptedException e) {
LOG.info("Server shutting down.");
@@ -272,8 +273,6 @@ public class Server {
}
public void shutdown() {
LOG.warn("Server shutdown 1");
List<ChannelFuture> channelFutures = new ArrayList<>();
if (tcpChannelHandle != null) {
@@ -306,8 +305,6 @@ public class Server {
}
});
LOG.warn("Server shutdown 2");
int quietPeriod = config.getServer().getShutdownQuietPeriod();
List<Future<?>> groupFutures = new ArrayList<>();
@@ -346,8 +343,6 @@ public class Server {
groupFutures.add(udpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
}
LOG.warn("Server shutdown 3");
if (udpWorkerGroup != null) {
LOG.info("Shutting down udpWorkerGroup");
groupFutures.add(udpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
@@ -361,17 +356,15 @@ public class Server {
}
});
LOG.warn("Server shutdown 4");
try {
LOG.info("Closing dataStore");
dataStore.close();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
if (dataStore != null) {
try {
LOG.info("Closing dataStore");
dataStore.close();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
}
LOG.warn("Server shutdown 5");
if (dataStoreCache != null) {
try {
LOG.info("Closing dataStoreCache");
@@ -381,8 +374,6 @@ public class Server {
}
}
LOG.warn("Server shutdown 6");
try {
LOG.info("Closing metaCacheFactory");
MetaCacheFactory.close();
@@ -390,8 +381,6 @@ public class Server {
LOG.error(e.getMessage(), e);
}
LOG.warn("Server shutdown 7");
try {
LOG.info("Closing webSocketRequestDecoder subscriptions");
WebSocketRequestDecoder.close();
@@ -399,8 +388,6 @@ public class Server {
LOG.error(e.getMessage(), e);
}
LOG.warn("Server shutdown 8");
if (curatorFramework != null) {
try {
LOG.info("Closing curatorFramework");
@@ -410,8 +397,6 @@ public class Server {
}
}
LOG.warn("Server shutdown 9");
if (applicationContext != null) {
try {
LOG.info("Closing applicationContext");
@@ -420,7 +405,6 @@ public class Server {
LOG.error(e.getMessage(), e);
}
}
LOG.warn("Server shutdown 10");
this.shutdown = true;
LOG.info("Server shut down.");
}
@@ -452,7 +436,7 @@ public class Server {
}
}
public void run() throws Exception {
public void run(SslContext sslCtx) throws Exception {
RetryPolicy retryPolicy = new RetryForever(1000);
int timeout = Long.valueOf(AccumuloConfiguration.getTimeInMillis(config.getAccumulo().getZookeeperTimeout()))
@@ -519,7 +503,6 @@ public class Server {
final int httpPort = config.getHttp().getPort();
final String httpIp = config.getHttp().getIp();
SslContext sslCtx = createSSLContext(config);
if (sslCtx instanceof OpenSslServerContext) {
OpenSslServerContext openssl = (OpenSslServerContext) sslCtx;
String application = "Timely_" + httpPort;
@@ -576,7 +559,7 @@ public class Server {
tcpAddress, tcpPort, httpAddress, httpPort, wsAddress, wsPort, wsAddress, udpPort);
}
protected SslContext createSSLContext(Configuration config) throws Exception {
public static SslContext createSSLContext(Configuration config) throws Exception {
ServerSsl sslCfg = config.getSecurity().getServerSsl();
Boolean generate = sslCfg.isUseGeneratedKeypair();
@@ -85,7 +85,7 @@ public class StandaloneServer extends Server {
try {
LOG.info("Starting StandaloneServer");
StandaloneServer s = new StandaloneServer(conf);
s.run();
s.run(createSSLContext(conf));
} catch (Exception e) {
System.err.println("Error starting server");
e.printStackTrace();
@@ -8,6 +8,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import com.google.common.base.Joiner;
@@ -39,6 +40,8 @@ public class MetricAdapter {
private static final TagParser tagParser = new TagParser();
private static final TagListParser tagListParser = new TagListParser();
private static Pattern equalsPattern = Pattern.compile("=");
private static Pattern commaPattern = Pattern.compile(",");
public static final ColumnVisibility EMPTY_VISIBILITY = new ColumnVisibility();
public static final String VISIBILITY_TAG = "viz";
@@ -59,10 +62,10 @@ public class MetricAdapter {
// are used as a delimiter
String k = t.getKey();
String v = t.getValue();
k = k.replaceAll("=", "\\=");
k = k.replaceAll(",", "\\,");
v = v.replaceAll("=", "\\=");
v = v.replaceAll(",", "\\,");
k = equalsPattern.matcher(k).replaceAll("\\=");
k = commaPattern.matcher(k).replaceAll("\\,");
v = equalsPattern.matcher(v).replaceAll("\\=");
v = commaPattern.matcher(v).replaceAll("\\,");
newTags.add(new Tag(k, v));
}
return newTags;
@@ -75,10 +78,10 @@ public class MetricAdapter {
// are used as a delimiter
String k = t.getKey();
String v = t.getValue();
k = k.replaceAll("=", "\\=");
k = k.replaceAll(",", "\\,");
v = v.replaceAll("=", "\\=");
v = v.replaceAll(",", "\\,");
k = equalsPattern.matcher(k).replaceAll("\\=");
k = commaPattern.matcher(k).replaceAll("\\,");
v = equalsPattern.matcher(v).replaceAll("\\=");
v = commaPattern.matcher(v).replaceAll("\\,");
newTags.put(k, v);
}
return newTags;
@@ -1,6 +1,7 @@
package timely.configuration;
import java.util.List;
import java.util.Objects;
import com.google.common.collect.Lists;
import timely.validator.NotEmptyIfFieldSet;
@@ -76,4 +77,24 @@ public class ServerSsl {
public void setUseCiphers(List<String> useCiphers) {
this.useCiphers = useCiphers;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
ServerSsl serverSsl = (ServerSsl) o;
return useGeneratedKeypair == serverSsl.useGeneratedKeypair && useOpenssl == serverSsl.useOpenssl
&& Objects.equals(certificateFile, serverSsl.certificateFile)
&& Objects.equals(keyFile, serverSsl.keyFile) && Objects.equals(keyPassword, serverSsl.keyPassword)
&& Objects.equals(trustStoreFile, serverSsl.trustStoreFile)
&& Objects.equals(useCiphers, serverSsl.useCiphers);
}
@Override
public int hashCode() {
return Objects.hash(certificateFile, keyFile, keyPassword, trustStoreFile, useGeneratedKeypair, useOpenssl,
useCiphers);
}
}
@@ -4,8 +4,15 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.data.ColumnUpdate;
@@ -251,6 +258,52 @@ public class DownsampleIteratorTest {
assertEquals((elapsedTime / sampleInterval) * numTagVariations, totalBuckets);
}
@Test
public void test() {
String hello = new String("this is a longer string");
long maybe = ObjectSizeOf.Sizer.OBJECT_OVERHEAD + // for String
ObjectSizeOf.Sizer.REFERENCE + // for array of char
ObjectSizeOf.Sizer.getPrimitiveObjectSize(int.class) +
ObjectSizeOf.Sizer.ARRAY_OVERHEAD + // for array of char
hello.length() * ObjectSizeOf.Sizer.getPrimitiveObjectSize(char.class);
maybe = ObjectSizeOf.Sizer.roundUp(maybe);
long size = ObjectSizeOf.Sizer.getObjectSize(hello, true, true);
System.out.println("size = " + size);
System.out.println("maybe = " + maybe);
}
@Test
public void test2() {
System.out.println(ObjectSizeOf.Sizer.getObjectSize(new Long(100)));
}
@Test
public void testDownsampleMemoryCalculation() throws Exception {
int numTagVariations = 1;
int sampleInterval = 50;
int elapsedTime = 100;
int skipInterval = 10;
SortedMap<Key, Value> testData3 = createTestData3(elapsedTime, skipInterval, numTagVariations);
DownsampleIterator iter = new DownsampleIterator();
Map<Set<Tag>, Downsample> samples = runQuery(iter, testData3, sampleInterval, 1000);
assertEquals(numTagVariations, samples.size());
long totalBuckets = 0;
for (Entry<Set<Tag>, Downsample> entry : samples.entrySet()) {
Set<Tag> tagSet = entry.getKey();
Downsample downsample = entry.getValue();
for (Tag tag : tagSet) {
assertEquals(ObjectSizeOf.Sizer.getObjectSize(tag, true, true),
ObjectSizeOf.Sizer.getObjectSize(tag, false, true));
}
}
}
private Map<Set<Tag>, Downsample> runQuery(SortedKeyValueIterator<Key, Value> iter, SortedMap<Key, Value> testData,
long period, long maxDownsampleMemory) throws Exception {
IteratorSetting is = new IteratorSetting(100, DownsampleIterator.class);
@@ -59,7 +59,7 @@ public class DataStoreIT extends OneWaySSLBase {
conf.setMetricAgeOffDays(ageOffSettings);
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + ONE_HOUR) + " 3.0 tag1=value1 tag2=value2",
@@ -106,7 +106,7 @@ public class DataStoreIT extends OneWaySSLBase {
conf.setMetricAgeOffDays(ageOffSettings);
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
// @formatter:off
/*
* TEST_TIME = System.currentTimeMillis() - ONE_DAY
@@ -5,6 +5,8 @@ import java.io.IOException;
import java.nio.file.Files;
import java.util.HashMap;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.accumulo.minicluster.MiniAccumuloConfig;
@@ -12,8 +14,10 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.Server;
import timely.auth.AuthCache;
import timely.configuration.Configuration;
import timely.configuration.ServerSsl;
import timely.store.MetricAgeOffIterator;
import timely.test.TestConfiguration;
@@ -24,6 +28,11 @@ public class MacITBase {
private static final Logger LOG = LoggerFactory.getLogger(MacITBase.class);
protected static ServerSsl serverSsl = null;
protected static File clientTrustStoreFile = null;
protected static SslContext sslCtx = null;
protected static SelfSignedCertificate serverCert = null;
private static final File tempDir;
protected static final String MAC_ROOT_USER = "root";
@@ -50,8 +59,6 @@ public class MacITBase {
conf = TestConfiguration.createMinimalConfigurationForTest();
conf.getAccumulo().setInstanceName(mac.getInstanceName());
conf.getAccumulo().setZookeepers(mac.getZooKeepers());
conf.getSecurity().getServerSsl().setUseOpenssl(false);
conf.getSecurity().getServerSsl().setUseGeneratedKeypair(true);
conf.getWebsocket().setFlushIntervalSeconds(TestConfiguration.WAIT_SECONDS);
HashMap<String, Integer> ageOffSettings = new HashMap<>();
ageOffSettings.put(MetricAgeOffIterator.DEFAULT_AGEOFF_KEY, 7);
@@ -88,4 +95,52 @@ public class MacITBase {
public void initializeAuthCache() {
AuthCache.clear();
}
protected static void setupSSL() throws Exception {
setupSSL(TestConfiguration.createMinimalConfigurationForTest(), false);
}
protected static void setupSSL(Configuration config, boolean twoWaySsl) throws Exception {
setupSSL(config, twoWaySsl, null);
}
protected static void setupSSL(Configuration config, boolean twoWaySsl, SelfSignedCertificate serverCert)
throws Exception {
ServerSsl serverSsl = config.getSecurity().getServerSsl();
boolean generateNewSsl = MacITBase.serverSsl == null || serverCert != null
|| !MacITBase.serverSsl.equals(serverSsl);
if (generateNewSsl) {
// new serverCert and trustStoreFile
if (serverCert == null) {
MacITBase.serverCert = new SelfSignedCertificate();
} else {
MacITBase.serverCert = serverCert;
}
MacITBase.clientTrustStoreFile = MacITBase.serverCert.certificate().getAbsoluteFile();
MacITBase.serverSsl = serverSsl;
MacITBase.serverSsl.setCertificateFile(MacITBase.serverCert.certificate().getAbsolutePath());
MacITBase.serverSsl.setKeyFile(MacITBase.serverCert.privateKey().getAbsolutePath());
}
config.getSecurity().getServerSsl().setCertificateFile(MacITBase.serverSsl.getCertificateFile());
config.getSecurity().getServerSsl().setKeyFile(MacITBase.serverSsl.getKeyFile());
config.getSecurity().getServerSsl().setUseOpenssl(false);
config.getSecurity().getServerSsl().setUseGeneratedKeypair(false);
if (twoWaySsl) {
// Needed for 2way SSL
config.getSecurity().getServerSsl().setTrustStoreFile(MacITBase.serverSsl.getCertificateFile());
config.getSecurity().setAllowAnonymousHttpAccess(false);
} else {
config.getSecurity().setAllowAnonymousHttpAccess(true);
}
if (generateNewSsl) {
// new SslContext
OneWaySSLBase.sslCtx = Server.createSSLContext(config);
}
}
protected static SslContext getSslContext() {
return MacITBase.sslCtx;
}
}
@@ -30,7 +30,7 @@ public class OneWaySSLAnonAccessIT extends OneWaySSLBase {
@Test
public void testQueryWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1000) + " 3.0 tag1=value1 tag2=value2",
@@ -1,6 +1,5 @@
package timely.test.integration;
import java.io.File;
import java.net.URL;
import javax.net.ssl.HttpsURLConnection;
@@ -12,25 +11,21 @@ import io.netty.handler.ssl.JdkSslContext;
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 org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import timely.auth.AuthCache;
import timely.configuration.Configuration;
/**
* Base test class for SSL with anonymous access
*/
public class OneWaySSLBase extends QueryBase {
protected static File clientTrustStoreFile = null;
protected SSLSocketFactory getSSLSocketFactory() throws Exception {
SslContextBuilder builder = SslContextBuilder.forClient();
builder.applicationProtocolConfig(ApplicationProtocolConfig.DISABLED);
builder.sslProvider(SslProvider.JDK);
builder.trustManager(clientTrustStoreFile); // Trust the server cert
builder.trustManager(MacITBase.clientTrustStoreFile); // Trust the server cert
SslContext ctx = builder.build();
Assert.assertTrue(ctx.isClient());
Assert.assertTrue(ctx instanceof JdkSslContext);
@@ -39,19 +34,9 @@ public class OneWaySSLBase extends QueryBase {
return jdkSslContext.getSocketFactory();
}
protected static void setupSSL(Configuration config) throws Exception {
SelfSignedCertificate serverCert = new SelfSignedCertificate();
config.getSecurity().getServerSsl().setCertificateFile(serverCert.certificate().getAbsolutePath());
clientTrustStoreFile = serverCert.certificate().getAbsoluteFile();
config.getSecurity().getServerSsl().setKeyFile(serverCert.privateKey().getAbsolutePath());
config.getSecurity().getServerSsl().setUseOpenssl(false);
config.getSecurity().getServerSsl().setUseGeneratedKeypair(false);
config.getSecurity().setAllowAnonymousHttpAccess(true);
}
@Before
public void configureSSL() throws Exception {
setupSSL(conf);
setupSSL(conf, false);
}
@Override
@@ -93,7 +93,7 @@ public class OneWaySSLBasicAuthAccessIT extends OneWaySSLBase {
@Test
public void testBasicAuthLogin() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
String metrics = "https://localhost:54322/api/metrics";
query(metrics);
@@ -105,7 +105,7 @@ public class OneWaySSLBasicAuthAccessIT extends OneWaySSLBase {
@Test(expected = UnauthorizedUserException.class)
public void testBasicAuthLoginFailure() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
String metrics = "https://localhost:54322/api/metrics";
query("test", "test2", metrics);
@@ -117,7 +117,7 @@ public class OneWaySSLBasicAuthAccessIT extends OneWaySSLBase {
@Test
public void testQueryWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1000) + " 3.0 tag1=value1 tag2=value2",
@@ -13,7 +13,7 @@ public class StaticFileServerIT extends OneWaySSLBase {
@Before
public void before() throws Exception {
s = new Server(conf);
s.run();
s.run(getSslContext());
}
@After
@@ -0,0 +1,92 @@
package timely.test.integration;
import java.net.URL;
import java.util.List;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import javax.net.ssl.SSLSocketFactory;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.cookie.ClientCookieDecoder;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.JdkSslContext;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslProvider;
import org.junit.After;
import org.junit.Assert;
import timely.auth.AuthCache;
import timely.netty.Constants;
/**
* Base test class for SSL with anonymous access
*/
public class TwoWaySSLBase extends QueryBase {
@After
public void tearDown() throws Exception {
AuthCache.resetConfiguration();
}
protected SSLSocketFactory getSSLSocketFactory() throws Exception {
SslContextBuilder builder = SslContextBuilder.forClient();
builder.applicationProtocolConfig(ApplicationProtocolConfig.DISABLED);
// Use server cert / key on client side
builder.keyManager(MacITBase.serverCert.key(), (String) null, MacITBase.serverCert.cert());
builder.sslProvider(SslProvider.JDK);
builder.trustManager(MacITBase.clientTrustStoreFile); // Trust the server cert
SslContext ctx = builder.build();
Assert.assertTrue(ctx.isClient());
Assert.assertTrue(ctx instanceof JdkSslContext);
JdkSslContext jdk = (JdkSslContext) ctx;
SSLContext jdkSslContext = jdk.context();
return jdkSslContext.getSocketFactory();
}
@Override
protected HttpsURLConnection getUrlConnection(URL url) throws Exception {
// Username and password not used in 2way SSL case
return getUrlConnection(null, null, url);
}
@Override
protected HttpsURLConnection getUrlConnection(String username, String password, URL url) throws Exception {
HttpsURLConnection.setDefaultSSLSocketFactory(getSSLSocketFactory());
URL loginURL = new URL(url.getProtocol() + "://" + url.getHost() + ":" + url.getPort() + "/login");
HttpsURLConnection con = (HttpsURLConnection) loginURL.openConnection();
con.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String arg0, SSLSession arg1) {
return true;
}
});
con.setRequestMethod("GET");
con.setDoOutput(true);
con.setRequestProperty("Content-Type", "application/json");
con.connect();
int responseCode = con.getResponseCode();
if (401 == responseCode) {
throw new UnauthorizedUserException();
}
Assert.assertEquals(200, responseCode);
List<String> cookies = con.getHeaderFields().get(HttpHeaderNames.SET_COOKIE.toString());
Assert.assertEquals(1, cookies.size());
Cookie sessionCookie = ClientCookieDecoder.STRICT.decode(cookies.get(0));
Assert.assertEquals(Constants.COOKIE_NAME, sessionCookie.name());
con = (HttpsURLConnection) url.openConnection();
con.setRequestProperty(HttpHeaderNames.COOKIE.toString(), sessionCookie.name() + "=" + sessionCookie.value());
con.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String arg0, SSLSession arg1) {
return true;
}
});
return con;
}
}
@@ -1,30 +1,11 @@
package timely.test.integration;
import java.io.File;
import java.net.URL;
import java.util.List;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import javax.net.ssl.SSLSocketFactory;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.cookie.ClientCookieDecoder;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.JdkSslContext;
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 org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.accumulo.minicluster.MiniAccumuloConfig;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -33,10 +14,9 @@ import org.junit.rules.TemporaryFolder;
import timely.Server;
import timely.auth.AuthCache;
import timely.configuration.Configuration;
import timely.netty.Constants;
import timely.test.TestConfiguration;
public class TwoWaySSLFailureIT extends QueryBase {
public class TwoWaySSLFailureIT extends TwoWaySSLBase {
@ClassRule
public static final TemporaryFolder temp = new TemporaryFolder();
@@ -44,87 +24,6 @@ public class TwoWaySSLFailureIT extends QueryBase {
private static MiniAccumuloCluster mac = null;
private static Configuration conf = null;
private static SelfSignedCertificate serverCert = null;
private static File clientTrustStoreFile = null;
static {
try {
// This fqdn does not match what is in security.xml
serverCert = new SelfSignedCertificate("CN=bad.example.com");
clientTrustStoreFile = serverCert.certificate().getAbsoluteFile();
} catch (Exception e) {
throw new RuntimeException("Error creating self signed certificate", e);
}
}
protected SSLSocketFactory getSSLSocketFactory() throws Exception {
SslContextBuilder builder = SslContextBuilder.forClient();
builder.applicationProtocolConfig(ApplicationProtocolConfig.DISABLED);
// Use server cert / key on client side
builder.keyManager(serverCert.key(), (String) null, serverCert.cert());
builder.sslProvider(SslProvider.JDK);
builder.trustManager(clientTrustStoreFile); // Trust the server cert
SslContext ctx = builder.build();
Assert.assertTrue(ctx.isClient());
Assert.assertTrue(ctx instanceof JdkSslContext);
JdkSslContext jdk = (JdkSslContext) ctx;
SSLContext jdkSslContext = jdk.context();
return jdkSslContext.getSocketFactory();
}
protected static void setupSSL(Configuration config) throws Exception {
config.getSecurity().getServerSsl().setCertificateFile(serverCert.certificate().getAbsolutePath());
config.getSecurity().getServerSsl().setKeyFile(serverCert.privateKey().getAbsolutePath());
// Needed for 2way SSL
config.getSecurity().getServerSsl().setTrustStoreFile(serverCert.certificate().getAbsolutePath());
config.getSecurity().getServerSsl().setUseOpenssl(false);
config.getSecurity().getServerSsl().setUseGeneratedKeypair(false);
config.getSecurity().setAllowAnonymousHttpAccess(false);
}
@Override
protected HttpsURLConnection getUrlConnection(URL url) throws Exception {
// Username and password not used in 2way SSL case
return getUrlConnection(null, null, url);
}
@Override
protected HttpsURLConnection getUrlConnection(String username, String password, URL url) throws Exception {
HttpsURLConnection.setDefaultSSLSocketFactory(getSSLSocketFactory());
URL loginURL = new URL(url.getProtocol() + "://" + url.getHost() + ":" + url.getPort() + "/login");
HttpsURLConnection con = (HttpsURLConnection) loginURL.openConnection();
con.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String arg0, SSLSession arg1) {
return true;
}
});
con.setRequestMethod("GET");
con.setDoOutput(true);
con.setRequestProperty("Content-Type", "application/json");
con.connect();
int responseCode = con.getResponseCode();
if (401 == responseCode) {
throw new UnauthorizedUserException();
}
Assert.assertEquals(200, responseCode);
List<String> cookies = con.getHeaderFields().get(HttpHeaderNames.SET_COOKIE.toString());
Assert.assertEquals(1, cookies.size());
Cookie sessionCookie = ClientCookieDecoder.STRICT.decode(cookies.get(0));
Assert.assertEquals(Constants.COOKIE_NAME, sessionCookie.name());
con = (HttpsURLConnection) url.openConnection();
con.setRequestProperty(HttpHeaderNames.COOKIE.toString(), sessionCookie.name() + "=" + sessionCookie.value());
con.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String arg0, SSLSession arg1) {
return true;
}
});
return con;
}
@BeforeClass
public static void beforeClass() throws Exception {
temp.create();
@@ -134,7 +33,8 @@ public class TwoWaySSLFailureIT extends QueryBase {
conf = TestConfiguration.createMinimalConfigurationForTest();
conf.getAccumulo().setInstanceName(mac.getInstanceName());
conf.getAccumulo().setZookeepers(mac.getZooKeepers());
setupSSL(conf);
// This fqdn does not match what is in security.xml
setupSSL(conf, true, new SelfSignedCertificate("CN=bad.example.com"));
}
@AfterClass
@@ -163,7 +63,7 @@ public class TwoWaySSLFailureIT extends QueryBase {
@Test(expected = UnauthorizedUserException.class)
public void testBasicAuthLoginFailure() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
String metrics = "https://localhost:54322/api/metrics";
query(metrics);
@@ -3,33 +3,17 @@ package timely.test.integration;
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static org.junit.Assert.assertEquals;
import java.io.File;
import java.net.URL;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.cookie.ClientCookieDecoder;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.JdkSslContext;
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 org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.accumulo.minicluster.MiniAccumuloConfig;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -41,7 +25,6 @@ import timely.api.request.timeseries.QueryRequest;
import timely.api.response.timeseries.QueryResponse;
import timely.auth.AuthCache;
import timely.configuration.Configuration;
import timely.netty.Constants;
import timely.test.IntegrationTest;
import timely.test.TestConfiguration;
@@ -51,7 +34,7 @@ import timely.test.TestConfiguration;
*
*/
@Category(IntegrationTest.class)
public class TwoWaySSLIT extends QueryBase {
public class TwoWaySSLIT extends TwoWaySSLBase {
private static final Long TEST_TIME = System.currentTimeMillis();
@@ -61,74 +44,6 @@ public class TwoWaySSLIT extends QueryBase {
private static MiniAccumuloCluster mac = null;
private static Configuration conf = null;
protected static SelfSignedCertificate serverCert = null;
protected static File clientTrustStoreFile = null;
static {
try {
serverCert = new SelfSignedCertificate();
clientTrustStoreFile = serverCert.certificate().getAbsoluteFile();
} catch (Exception e) {
throw new RuntimeException("Error creating self signed certificate", e);
}
}
protected SSLSocketFactory getSSLSocketFactory() throws Exception {
SslContextBuilder builder = SslContextBuilder.forClient();
builder.applicationProtocolConfig(ApplicationProtocolConfig.DISABLED);
// Use server cert / key on client side.
builder.keyManager(serverCert.key(), (String) null, serverCert.cert());
builder.sslProvider(SslProvider.JDK);
builder.trustManager(clientTrustStoreFile); // Trust the server cert
SslContext ctx = builder.build();
Assert.assertTrue(ctx.isClient());
Assert.assertTrue(ctx instanceof JdkSslContext);
JdkSslContext jdk = (JdkSslContext) ctx;
SSLContext jdkSslContext = jdk.context();
return jdkSslContext.getSocketFactory();
}
protected static void setupSSL(Configuration config) throws Exception {
config.getSecurity().getServerSsl().setCertificateFile(serverCert.certificate().getAbsolutePath());
config.getSecurity().getServerSsl().setKeyFile(serverCert.privateKey().getAbsolutePath());
// Needed for 2way SSL
config.getSecurity().getServerSsl().setTrustStoreFile(serverCert.certificate().getAbsolutePath());
config.getSecurity().getServerSsl().setUseOpenssl(false);
config.getSecurity().getServerSsl().setUseGeneratedKeypair(false);
config.getSecurity().setAllowAnonymousHttpAccess(false);
}
@Override
protected HttpsURLConnection getUrlConnection(URL url) throws Exception {
// Username and password not used in 2way SSL case
return getUrlConnection(null, null, url);
}
@Override
protected HttpsURLConnection getUrlConnection(String username, String password, URL url) throws Exception {
HttpsURLConnection.setDefaultSSLSocketFactory(getSSLSocketFactory());
URL loginURL = new URL(url.getProtocol() + "://" + url.getHost() + ":" + url.getPort() + "/login");
HttpsURLConnection con = (HttpsURLConnection) loginURL.openConnection();
con.setHostnameVerifier((host, session) -> true);
con.setRequestMethod("GET");
con.setDoOutput(true);
con.setRequestProperty("Content-Type", "application/json");
con.connect();
int responseCode = con.getResponseCode();
if (401 == responseCode) {
throw new UnauthorizedUserException();
}
Assert.assertEquals(200, responseCode);
List<String> cookies = con.getHeaderFields().get(HttpHeaderNames.SET_COOKIE.toString());
Assert.assertEquals(1, cookies.size());
Cookie sessionCookie = ClientCookieDecoder.STRICT.decode(cookies.get(0));
Assert.assertEquals(Constants.COOKIE_NAME, sessionCookie.name());
con = (HttpsURLConnection) url.openConnection();
con.setRequestProperty(HttpHeaderNames.COOKIE.toString(), sessionCookie.name() + "=" + sessionCookie.value());
con.setHostnameVerifier((host, session) -> true);
return con;
}
@BeforeClass
public static void beforeClass() throws Exception {
final MiniAccumuloConfig macConfig = new MiniAccumuloConfig(temp.newFolder("mac"), "secret");
@@ -137,7 +52,6 @@ public class TwoWaySSLIT extends QueryBase {
conf = TestConfiguration.createMinimalConfigurationForTest();
conf.getAccumulo().setInstanceName(mac.getInstanceName());
conf.getAccumulo().setZookeepers(mac.getZooKeepers());
setupSSL(conf);
}
@AfterClass
@@ -157,6 +71,7 @@ public class TwoWaySSLIT extends QueryBase {
}
}
});
setupSSL(conf, true);
}
@After
@@ -167,7 +82,7 @@ public class TwoWaySSLIT extends QueryBase {
@Test
public void testBasicAuthLogin() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
String metrics = "https://localhost:54322/api/metrics";
query(metrics);
@@ -190,7 +105,7 @@ public class TwoWaySSLIT extends QueryBase {
public void testQueryWithVisibility(Configuration conf) throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
// @formatter:off
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
@@ -3,32 +3,14 @@ package timely.test.integration;
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static org.junit.Assert.assertEquals;
import java.io.File;
import java.net.URL;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import javax.net.ssl.SSLSocketFactory;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.cookie.ClientCookieDecoder;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.JdkSslContext;
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 org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.security.Authorizations;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -36,8 +18,6 @@ import timely.Server;
import timely.api.request.timeseries.QueryRequest;
import timely.api.response.timeseries.QueryResponse;
import timely.auth.AuthCache;
import timely.configuration.Configuration;
import timely.netty.Constants;
import timely.test.IntegrationTest;
import timely.test.TestConfiguration;
@@ -48,97 +28,13 @@ import timely.test.TestConfiguration;
*
*/
@Category(IntegrationTest.class)
public class TwoWaySSLOpenSSLIT extends QueryBase {
public class TwoWaySSLOpenSSLIT extends TwoWaySSLBase {
private static final Long TEST_TIME = System.currentTimeMillis();
protected static SelfSignedCertificate serverCert = null;
protected static File clientTrustStoreFile = null;
static {
try {
serverCert = new SelfSignedCertificate();
clientTrustStoreFile = serverCert.certificate().getAbsoluteFile();
} catch (Exception e) {
throw new RuntimeException("Error creating self signed certificate", e);
}
}
protected SSLSocketFactory getSSLSocketFactory() throws Exception {
SslContextBuilder builder = SslContextBuilder.forClient();
builder.applicationProtocolConfig(ApplicationProtocolConfig.DISABLED);
// Use server cert / key on client side.
builder.keyManager(serverCert.key(), (String) null, serverCert.cert());
builder.sslProvider(SslProvider.JDK);
builder.trustManager(clientTrustStoreFile); // Trust the server cert
SslContext ctx = builder.build();
Assert.assertTrue(ctx.isClient());
Assert.assertTrue(ctx instanceof JdkSslContext);
JdkSslContext jdk = (JdkSslContext) ctx;
SSLContext jdkSslContext = jdk.context();
return jdkSslContext.getSocketFactory();
}
protected static void setupSSL(Configuration config) throws Exception {
config.getSecurity().getServerSsl().setCertificateFile(serverCert.certificate().getAbsolutePath());
config.getSecurity().getServerSsl().setKeyFile(serverCert.privateKey().getAbsolutePath());
// Needed for 2way SSL
config.getSecurity().getServerSsl().setTrustStoreFile(serverCert.certificate().getAbsolutePath());
config.getSecurity().getServerSsl().setUseOpenssl(false);
config.getSecurity().getServerSsl().setUseGeneratedKeypair(false);
config.getSecurity().setAllowAnonymousHttpAccess(false);
}
@Before
public void configureSSL() throws Exception {
setupSSL(conf);
}
@Override
protected HttpsURLConnection getUrlConnection(URL url) throws Exception {
// Username and password not used in 2way SSL case
return getUrlConnection(null, null, url);
}
@Override
protected HttpsURLConnection getUrlConnection(String username, String password, URL url) throws Exception {
HttpsURLConnection.setDefaultSSLSocketFactory(getSSLSocketFactory());
URL loginURL = new URL(url.getProtocol() + "://" + url.getHost() + ":" + url.getPort() + "/login");
HttpsURLConnection con = (HttpsURLConnection) loginURL.openConnection();
con.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String arg0, SSLSession arg1) {
return true;
}
});
con.setRequestMethod("GET");
con.setDoOutput(true);
con.setRequestProperty("Content-Type", "application/json");
con.connect();
int responseCode = con.getResponseCode();
if (401 == responseCode) {
throw new UnauthorizedUserException();
}
Assert.assertEquals(200, responseCode);
List<String> cookies = con.getHeaderFields().get(HttpHeaderNames.SET_COOKIE.toString());
Assert.assertEquals(1, cookies.size());
Cookie sessionCookie = ClientCookieDecoder.STRICT.decode(cookies.get(0));
Assert.assertEquals(Constants.COOKIE_NAME, sessionCookie.name());
con = (HttpsURLConnection) url.openConnection();
con.setRequestProperty(HttpHeaderNames.COOKIE.toString(), sessionCookie.name() + "=" + sessionCookie.value());
con.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String arg0, SSLSession arg1) {
return true;
}
});
return con;
}
@Before
public void setup() throws Exception {
setupSSL(conf, true);
Connector con = mac.getConnector("root", "secret");
con.securityOperations().changeUserAuthorizations("root", new Authorizations("A", "B", "C", "D", "E", "F"));
}
@@ -151,7 +47,7 @@ public class TwoWaySSLOpenSSLIT extends QueryBase {
@Test
public void testBasicAuthLogin() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
String metrics = "https://localhost:54322/api/metrics";
query(metrics);
@@ -163,7 +59,7 @@ public class TwoWaySSLOpenSSLIT extends QueryBase {
@Test
public void testQueryWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1000) + " 3.0 tag1=value1 tag2=value2",
@@ -49,7 +49,7 @@ public class TcpClientIT extends OneWaySSLBase {
@Test
public void testPut() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (TcpClient client = new TcpClient("127.0.0.1", 54321)) {
client.open();
client.write("put sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2\n");
@@ -80,7 +80,7 @@ public class TcpClientIT extends OneWaySSLBase {
public void testPutMultiple() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (TcpClient client = new TcpClient("127.0.0.1", 54321)) {
client.open();
// @formatter:off
@@ -124,7 +124,7 @@ public class TcpClientIT extends OneWaySSLBase {
@Test
public void testPutInvalidTimestamp() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (TcpClient client = new TcpClient("127.0.0.1", 54321)) {
client.open();
client.write("put sys.cpu.user " + TEST_TIME + "Z" + " 1.0 tag1=value1 tag2=value2\n");
@@ -139,7 +139,7 @@ public class TcpClientIT extends OneWaySSLBase {
@Test
public void testPersistence() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4",
@@ -183,7 +183,7 @@ public class TcpClientIT extends OneWaySSLBase {
@Test
public void testPersistenceWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4 viz=(a|b)",
@@ -53,6 +53,7 @@ public class UdpClientIT extends MacITBase {
}
}
});
setupSSL();
}
@After
@@ -63,7 +64,7 @@ public class UdpClientIT extends MacITBase {
@Test
public void testPut() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (UdpClient client = new UdpClient("127.0.0.1", 54325)) {
client.open();
client.write(("put sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2\n"));
@@ -93,7 +94,7 @@ public class UdpClientIT extends MacITBase {
public void testPutMultiple() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (UdpClient client = new UdpClient("127.0.0.1", 54325)) {
client.open();
client.write(("put sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2\n" + "put sys.cpu.idle "
@@ -135,7 +136,7 @@ public class UdpClientIT extends MacITBase {
@Test
public void testPutInvalidTimestamp() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (UdpClient client = new UdpClient("127.0.0.1", 54325)) {
client.open();
client.write(("put sys.cpu.user " + TEST_TIME + "Z" + " 1.0 tag1=value1 tag2=value2\n"));
@@ -149,7 +150,7 @@ public class UdpClientIT extends MacITBase {
@Test
public void testPersistence() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4",
@@ -193,7 +194,7 @@ public class UdpClientIT extends MacITBase {
@Test
public void testPersistenceWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4 viz=(a|b)",
@@ -65,7 +65,7 @@ public class WebSocketClientIT extends OneWaySSLBase {
Connector con = mac.getConnector("root", "secret");
con.securityOperations().changeUserAuthorizations("root", new Authorizations("A", "B", "C", "D", "E", "F"));
s = new Server(conf);
s.run();
s.run(getSslContext());
setupSslCtx();
}
@@ -23,7 +23,7 @@ public class HTTPStrictTransportSecurityIT extends OneWaySSLBase {
@Before
public void before() throws Exception {
s = new Server(conf);
s.run();
s.run(getSslContext());
}
@After
@@ -58,7 +58,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testSuggest() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4",
@@ -86,7 +86,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testMetrics() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
// @formatter:off
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
@@ -120,7 +120,7 @@ public class HttpApiIT extends OneWaySSLBase {
// @formatter:off
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4",
@@ -175,7 +175,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testLookup() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3",
@@ -211,7 +211,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithMsResolution() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3",
@@ -259,7 +259,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithoutMsResolution() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3",
@@ -298,7 +298,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithNoTags() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3",
@@ -334,7 +334,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithNoTagsMultipleSeries() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 host=h1",
"sys.cpu.user " + TEST_TIME + " 2.0 tag1=value1 tag2=value2 host=h2",
@@ -375,7 +375,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test(expected = NotSuccessfulException.class)
public void testQueryWithNoMatchingTags() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1",
"sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3 rack=r2",
@@ -399,7 +399,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithTagWildcard() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
// @formatter:off
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1",
@@ -464,7 +464,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithTagWildcard2() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
// @formatter:off
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1",
@@ -524,7 +524,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithTagOr() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
// @formatter:off
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1",
@@ -608,7 +608,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithTagRegex() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
System.out.println("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1");
System.out.println("sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3 rack=r2");
@@ -674,7 +674,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testQueryWithTagRegex2() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1",
"sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3 rack=r2",
@@ -730,7 +730,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testRateQuery() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
// @formatter:off
put("sys.cpu.user " + (TEST_TIME) + " 1.0 tag1=value1 tag2=value2 rack=r1",
@@ -782,7 +782,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testRateCounterQuery() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
// @formatter:off
put("sys.cpu.user " + (TEST_TIME) + " 1.0 tag1=value1 tag2=value2 rack=r1",
@@ -836,7 +836,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testRateCounterWithOptions() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
// @formatter:off
put("sys.cpu.user " + (TEST_TIME) + " 1.0 tag1=value1 tag2=value2 rack=r1",
@@ -892,7 +892,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testGetVersion() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
String response = query("https://127.0.0.1:54322/version", "application/json");
assertNotNull(response);
@@ -905,7 +905,7 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testPutMetric() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
Metric m = Metric.newBuilder().name("sys.cpu.user").value(TEST_TIME, 1.0D).tag(new Tag("tag1", "value1"))
.build();
@@ -56,7 +56,7 @@ public class TimelyTcpIT extends MacITBase {
@Test
public void testVersion() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (Socket sock = new Socket("127.0.0.1", 54321);
PrintWriter writer = new PrintWriter(sock.getOutputStream(), true);) {
writer.write("version\n");
@@ -76,7 +76,7 @@ public class TimelyTcpIT extends MacITBase {
@Test
public void testPut() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (Socket sock = new Socket("127.0.0.1", 54321);
PrintWriter writer = new PrintWriter(sock.getOutputStream(), true);) {
writer.write("put sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2\n");
@@ -107,7 +107,7 @@ public class TimelyTcpIT extends MacITBase {
public void testPutMultiple() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (Socket sock = new Socket("127.0.0.1", 54321);
PrintWriter writer = new PrintWriter(sock.getOutputStream(), true)) {
// @formatter:off
@@ -190,7 +190,7 @@ public class TimelyTcpIT extends MacITBase {
LOG.debug("Sending {} bytes", data.length);
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (Socket sock = new Socket("127.0.0.1", 54321);) {
sock.getOutputStream().write(data);
sock.getOutputStream().flush();
@@ -233,7 +233,7 @@ public class TimelyTcpIT extends MacITBase {
@Test
public void testPutInvalidTimestamp() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
try (Socket sock = new Socket("127.0.0.1", 54321);
PrintWriter writer = new PrintWriter(sock.getOutputStream(), true);
BufferedReader reader = new BufferedReader(new InputStreamReader(sock.getInputStream()));) {
@@ -249,7 +249,7 @@ public class TimelyTcpIT extends MacITBase {
@Test
public void testPersistence() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4",
@@ -293,7 +293,7 @@ public class TimelyTcpIT extends MacITBase {
@Test
public void testPersistenceWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4 viz=(a|b)",
@@ -60,6 +60,7 @@ public class TimelyUdpIT extends MacITBase {
}
}
});
setupSSL();
}
@After
@@ -70,7 +71,7 @@ public class TimelyUdpIT extends MacITBase {
@Test
public void testPut() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 54325);
DatagramPacket packet = new DatagramPacket("".getBytes(UTF_8), 0, 0, address.getAddress(), 54325);
try (DatagramSocket sock = new DatagramSocket()) {
@@ -102,7 +103,7 @@ public class TimelyUdpIT extends MacITBase {
public void testPutMultiple() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 54325);
DatagramPacket packet = new DatagramPacket("".getBytes(UTF_8), 0, 0, address.getAddress(), 54325);
// @formatter:off
@@ -185,7 +186,7 @@ public class TimelyUdpIT extends MacITBase {
LOG.debug("Sending {} bytes", data.length);
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 54325);
DatagramPacket packet = new DatagramPacket("".getBytes(UTF_8), 0, 0, address.getAddress(), 54325);
try (DatagramSocket sock = new DatagramSocket()) {
@@ -229,7 +230,7 @@ public class TimelyUdpIT extends MacITBase {
@Test
public void testPutInvalidTimestamp() throws Exception {
final TestServer m = new TestServer(conf);
m.run();
m.run(getSslContext());
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 54325);
DatagramPacket packet = new DatagramPacket("".getBytes(UTF_8), 0, 0, address.getAddress(), 54325);
try (DatagramSocket sock = new DatagramSocket();) {
@@ -245,7 +246,7 @@ public class TimelyUdpIT extends MacITBase {
@Test
public void testPersistence() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4",
@@ -289,7 +290,7 @@ public class TimelyUdpIT extends MacITBase {
@Test
public void testPersistenceWithVisibility() throws Exception {
final Server s = new Server(conf);
s.run();
s.run(getSslContext());
try {
put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
"sys.cpu.idle " + (TEST_TIME + 1) + " 1.0 tag3=value3 tag4=value4 viz=(a|b)",
@@ -189,7 +189,7 @@ public class WebSocketIT extends OneWaySSLBase {
@Before
public void setup() throws Exception {
s = new Server(conf);
s.run();
s.run(getSslContext());
Connector con = mac.getConnector("root", "secret");
con.securityOperations().changeUserAuthorizations("root", new Authorizations("A", "B", "C", "D", "E", "F"));