Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ public interface QueryServices extends SQLCloseable {
public static final String STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB = "phoenix.stats.guidepost.width";
public static final String STATS_GUIDEPOST_PER_REGION_ATTRIB = "phoenix.stats.guidepost.per.region";
public static final String STATS_USE_CURRENT_TIME_ATTRIB = "phoenix.stats.useCurrentTime";

@Deprecated // use STATS_COLLECTION_ENABLED config instead
public static final String STATS_ENABLED_ATTRIB = "phoenix.stats.enabled";

Expand Down Expand Up @@ -271,6 +271,11 @@ public interface QueryServices extends SQLCloseable {
public static final String DEFAULT_COLUMN_ENCODED_BYTES_ATRRIB = "phoenix.default.column.encoded.bytes.attrib";
public static final String DEFAULT_IMMUTABLE_STORAGE_SCHEME_ATTRIB = "phoenix.default.immutable.storage.scheme";
public static final String DEFAULT_MULTITENANT_IMMUTABLE_STORAGE_SCHEME_ATTRIB = "phoenix.default.multitenant.immutable.storage.scheme";

public static final String PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH = "phoenix.queryserver.base.path";
public static final String PHOENIX_QUERY_SERVER_SERVICE_NAME = "phoenix.queryserver.service.name";
public static final String PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME = "phoenix.queryserver.zookeeper.acl.username";
public static final String PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD = "phoenix.queryserver.zookeeper.acl.password";
public static final String STATS_COLLECTION_ENABLED = "phoenix.stats.collection.enabled";
public static final String USE_STATS_FOR_PARALLELIZATION = "phoenix.use.stats.parallelization";

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,10 @@
import static org.apache.phoenix.query.QueryServices.THREAD_POOL_SIZE_ATTRIB;
import static org.apache.phoenix.query.QueryServices.THREAD_TIMEOUT_MS_ATTRIB;
import static org.apache.phoenix.query.QueryServices.TRACING_BATCH_SIZE;
import static org.apache.phoenix.query.QueryServices.PHOENIX_QUERY_SERVER_SERVICE_NAME;
import static org.apache.phoenix.query.QueryServices.PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH;
import static org.apache.phoenix.query.QueryServices.PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD;
import static org.apache.phoenix.query.QueryServices.PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME;
import static org.apache.phoenix.query.QueryServices.TRACING_ENABLED;
import static org.apache.phoenix.query.QueryServices.TRACING_STATS_TABLE_NAME_ATTRIB;
import static org.apache.phoenix.query.QueryServices.TRACING_THREAD_POOL_SIZE;
Expand Down Expand Up @@ -179,8 +183,8 @@ public class QueryServicesOptions {
public static final int DEFAULT_CLOCK_SKEW_INTERVAL = 2000;
public static final boolean DEFAULT_INDEX_FAILURE_HANDLING_REBUILD = true; // auto rebuild on
public static final boolean DEFAULT_INDEX_FAILURE_BLOCK_WRITE = false;
public static final boolean DEFAULT_INDEX_FAILURE_DISABLE_INDEX = true;
public static final boolean DEFAULT_INDEX_FAILURE_THROW_EXCEPTION = true;
public static final boolean DEFAULT_INDEX_FAILURE_DISABLE_INDEX = true;
public static final boolean DEFAULT_INDEX_FAILURE_THROW_EXCEPTION = true;
public static final long DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_INTERVAL = 60000; // 60 secs
public static final long DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_BACKWARD_TIME = 1; // 1 ms
public static final long DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_FORWARD_TIME = 60000 * 3; // 3 mins
Expand Down Expand Up @@ -295,6 +299,10 @@ public class QueryServicesOptions {
public static final int DEFAULT_COLUMN_ENCODED_BYTES = QualifierEncodingScheme.TWO_BYTE_QUALIFIERS.getSerializedMetadataValue();
public static final String DEFAULT_IMMUTABLE_STORAGE_SCHEME = ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS.toString();
public static final String DEFAULT_MULTITENANT_IMMUTABLE_STORAGE_SCHEME = ImmutableStorageScheme.ONE_CELL_PER_COLUMN.toString();
public final static String DEFAULT_PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH = "/phoenix";
public final static String DEFAULT_PHOENIX_QUERY_SERVER_SERVICE_NAME = "queryserver";
public final static String DEFAULT_PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME = "phoenix";
public final static String DEFAULT_PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD = "phoenix";

//by default, max connections from one client to one cluster is unlimited
public static final int DEFAULT_CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS = 0;
Expand Down Expand Up @@ -388,6 +396,10 @@ public static QueryServicesOptions withDefaults() {
.setIfUnset(IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE, DEFAULT_IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE)
.setIfUnset(LOCAL_INDEX_CLIENT_UPGRADE_ATTRIB, DEFAULT_LOCAL_INDEX_CLIENT_UPGRADE)
.setIfUnset(AUTO_UPGRADE_ENABLED, DEFAULT_AUTO_UPGRADE_ENABLED)
.setIfUnset(PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH, DEFAULT_PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH)
.setIfUnset(PHOENIX_QUERY_SERVER_SERVICE_NAME, DEFAULT_PHOENIX_QUERY_SERVER_SERVICE_NAME)
.setIfUnset(PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME, DEFAULT_PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME)
.setIfUnset(PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD, DEFAULT_PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD)
.setIfUnset(UPLOAD_BINARY_DATA_TYPE_ENCODING, DEFAULT_UPLOAD_BINARY_DATA_TYPE_ENCODING)
.setIfUnset(TRACING_ENABLED, DEFAULT_TRACING_ENABLED)
.setIfUnset(TRACING_BATCH_SIZE, DEFAULT_TRACING_BATCH_SIZE)
Expand Down
84 changes: 84 additions & 0 deletions phoenix-load-balancer/pom.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
<?xml version='1.0'?>
<!--

Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.

-->

<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix</artifactId>
<version>4.12.0-HBase-1.3-SNAPSHOT</version>
</parent>
<artifactId>phoenix-load-balancer</artifactId>
<name>Phoenix Load Balancer</name>
<description>A Load balancer which routes calls to Phoenix Query Server</description>

<dependencies>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like you're using a bunch of dependencies transitively. This will work for now, but is brittle in the long term.

We should make sure that this module depends directly on all of the dependencies that it uses instead of relying on another dependency bringing them in. Can do this later though.

</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-client</artifactId>
</dependency>
<dependency>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix-queryserver</artifactId>
</dependency>
<!-- for tests -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<artifactId>maven-source-plugin</artifactId>
<executions>
<execution>
<id>attach-sources</id>
<phase>verify</phase>
<goals>
<goal>jar-no-fork</goal>
<goal>test-jar-no-fork</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<exclude>src/main/resources/META-INF/services/org.apache.phoenix.loadbalancer.service.LoadBalanceZookeeperConf</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>

</project>
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
/**
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.phoenix.end2end;

import com.google.common.net.HostAndPort;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.CuratorZookeeperClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.TestingServer;
import org.apache.curator.utils.CloseableUtils;
import org.apache.phoenix.loadbalancer.service.LoadBalancer;
import org.apache.phoenix.loadbalancer.service.LoadBalanceZookeeperConf;
import org.apache.phoenix.loadbalancer.service.LoadBalanceZookeeperConfImpl;
import org.apache.phoenix.queryserver.register.Registry;
import org.apache.phoenix.queryserver.register.ZookeeperRegistry;
import org.apache.zookeeper.KeeperException;
import org.junit.*;

import java.util.Arrays;
import java.util.List;

public class LoadBalancerEnd2EndIT {
private static TestingServer testingServer;
private static CuratorFramework curatorFramework;
private static final Log LOG = LogFactory.getLog(LoadBalancerEnd2EndIT.class);
private static final LoadBalanceZookeeperConf LOAD_BALANCER_CONFIGURATION = new LoadBalanceZookeeperConfImpl();
private static String path;
private static LoadBalancer loadBalancer;
private static HostAndPort pqs1 = HostAndPort.fromParts("localhost",1000);
private static HostAndPort pqs2 = HostAndPort.fromParts("localhost",2000);
private static HostAndPort pqs3 = HostAndPort.fromParts("localhost",3000);
public static String zkConnectString;
public static Registry registry;

@BeforeClass
public static void setup() throws Exception{

registry = new ZookeeperRegistry();
zkConnectString = LOAD_BALANCER_CONFIGURATION.getZkConnectString();
int port = Integer.parseInt(zkConnectString.split(":")[1]);
testingServer = new TestingServer(port);
testingServer.start();

path = LOAD_BALANCER_CONFIGURATION.getParentPath();
curatorFramework = CuratorFrameworkFactory.newClient(zkConnectString,
new ExponentialBackoffRetry(1000, 3));
curatorFramework.start();
createNodeForTesting(Arrays.asList(pqs1,pqs2,pqs3));
curatorFramework.setACL().withACL(LOAD_BALANCER_CONFIGURATION.getAcls());
loadBalancer = LoadBalancer.getLoadBalancer();
}

@AfterClass
public static void tearDown() throws Exception {
CloseableUtils.closeQuietly(curatorFramework);
CloseableUtils.closeQuietly(testingServer);
}

private static void createNodeForTesting(List<HostAndPort> pqsNodes) throws Exception{
for(HostAndPort pqs:pqsNodes) {
registry.registerServer(LOAD_BALANCER_CONFIGURATION,pqs.getPort(),zkConnectString,pqs.getHostText());
}
curatorFramework.getChildren().forPath(LOAD_BALANCER_CONFIGURATION.getParentPath()).size();
}


@Test
public void testGetAllServiceLocation() throws Exception {
Assert.assertNotNull(loadBalancer);
List<HostAndPort> serviceLocations = loadBalancer.getAllServiceLocation();
Assert.assertTrue(" must contains 3 service location",serviceLocations.size() == 3);
}

@Test
public void testGetSingleServiceLocation() throws Exception {
Assert.assertNotNull(loadBalancer);
HostAndPort serviceLocation = loadBalancer.getSingleServiceLocation();
Assert.assertNotNull(serviceLocation);
}

@Test(expected=Exception.class)
public void testZookeeperDown() throws Exception{
testingServer.stop();
CuratorZookeeperClient zookeeperClient = curatorFramework.getZookeeperClient();
//check to see if zookeeper is really down.
while (zookeeperClient.isConnected()){
Thread.sleep(1000);
};
loadBalancer.getSingleServiceLocation();
}

@Test(expected = KeeperException.NoNodeException.class)
public void testNoPhoenixQueryServerNodeInZookeeper() throws Exception{
List<HostAndPort> hostAndPorts = Arrays.asList(pqs1, pqs2, pqs3);
for(HostAndPort pqs: hostAndPorts) {
String fullPathToNode = LOAD_BALANCER_CONFIGURATION.getFullPathToNode(pqs);
curatorFramework.delete().deletingChildrenIfNeeded().forPath(fullPathToNode);
while (curatorFramework.checkExists().forPath(fullPathToNode) != null){
//wait for the node to deleted
Thread.sleep(1000);
};
}
//delete the parent
curatorFramework.delete().forPath(path);
// should throw an exception as there is
// no node in the zookeeper
try {
loadBalancer.getSingleServiceLocation();
} catch(Exception e) {
throw e;
} finally {
// need to create node for other tests to run.
createNodeForTesting(hostAndPorts);
}
}

@Test
public void testSingletonPropertyForLoadBalancer(){
LoadBalancer anotherloadBalancerRef = LoadBalancer.getLoadBalancer();
Assert.assertTrue(" the load balancer is not singleton",loadBalancer == anotherloadBalancerRef );
}



}
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
/**
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.phoenix.loadbalancer.service;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.net.HostAndPort;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;

import java.util.Arrays;
import java.util.List;



public class LoadBalanceZookeeperConfImpl implements LoadBalanceZookeeperConf {

private Configuration configuration;

public LoadBalanceZookeeperConfImpl() {
this.configuration = HBaseConfiguration.create();
}

public LoadBalanceZookeeperConfImpl(Configuration configuration) {
this.configuration = configuration;
}

@VisibleForTesting
public void setConfiguration(Configuration configuration) {
this.configuration = configuration;
}

@Override
public String getQueryServerBasePath(){
return configuration.get(QueryServices.PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH,
QueryServicesOptions.DEFAULT_PHOENIX_QUERY_SERVER_CLUSTER_BASE_PATH);
}

@Override
public String getServiceName(){
return configuration.get(QueryServices.PHOENIX_QUERY_SERVER_SERVICE_NAME,
QueryServicesOptions.DEFAULT_PHOENIX_QUERY_SERVER_SERVICE_NAME);
}

@Override
public String getZkConnectString(){
return String.format("%s:%s",configuration.get(QueryServices.ZOOKEEPER_QUORUM_ATTRIB,
"localhost"),configuration.get(QueryServices.ZOOKEEPER_PORT_ATTRIB,"2181"));
}

private String getZkLbUserName(){
return configuration.get(QueryServices.PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME,
QueryServicesOptions.DEFAULT_PHOENIX_QUERY_SERVER_ZK_ACL_USERNAME);
}

private String getZkLbPassword(){
return configuration.get(QueryServices.PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD,
QueryServicesOptions.DEFAULT_PHOENIX_QUERY_SERVER_ZK_ACL_PASSWORD);
}

@Override
public List<ACL> getAcls() {
ACL acl = new ACL();
acl.setId(new Id("digest",getZkLbUserName()+":"+getZkLbPassword()));
acl.setPerms(ZooDefs.Perms.READ);
return Arrays.asList(acl);
}

@Override
public String getParentPath() {
String path = String.format("%s/%s",getQueryServerBasePath(),getServiceName());
return path;
}

@Override
public String getFullPathToNode(HostAndPort hostAndPort) {
String path = String.format("%s/%s",getParentPath()
,hostAndPort.toString());
return path;
}
}

Loading