feat(pd): integrate `pd-dist` into hugegraph & add core tests, client tests, and REST tests for PD (#2532)

subtask of #2265

---------

Co-authored-by: imbajin <jin@apache.org>
This commit is contained in:
V_Galaxy 2024-05-06 20:08:48 +08:00 committed by GitHub
parent 2b3bc4f28b
commit 253b8d38b8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
44 changed files with 4568 additions and 27 deletions

View File

@ -44,10 +44,31 @@ jobs:
cp $HOME/.m2/settings.xml /tmp/settings.xml
mv -vf .github/configs/settings.xml $HOME/.m2/settings.xml
- name: Package
run: |
mvn clean package -U -Dmaven.javadoc.skip=true -Dmaven.test.skip=true -ntp
- name: Run common test
run: |
mvn test -pl hugegraph-pd/hg-pd-test -am -P pd-common-test
- name: Run core test
run: |
mvn test -pl hugegraph-pd/hg-pd-test -am -P pd-core-test
# The above tests do not require starting a PD instance.
- name: Prepare env and service
run: |
$TRAVIS_DIR/start-pd.sh
- name: Run client test
run: |
mvn test -pl hugegraph-pd/hg-pd-test -am -P pd-client-test
- name: Run rest test
run: |
mvn test -pl hugegraph-pd/hg-pd-test -am -P pd-rest-test
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3.0.0
with:

View File

@ -73,8 +73,15 @@ public class ClientCache {
try {
GraphCache graph = initGraph(graphName);
Partition partition = graph.getPartition(partId);
Shard shard = groups.get(partId).getValue();
if (partition == null || shard == null) {
if (partition == null) {
return null;
}
KVPair<ShardGroup, Shard> group = groups.get(partId);
if (group == null) {
return null;
}
Shard shard = group.getValue();
if (shard == null) {
return null;
}
return new KVPair<>(partition, shard);

View File

@ -46,7 +46,7 @@ public class IdMetaStore extends MetadataRocksDBStore {
private static final String CID_DEL_SLOT_PREFIX = "@CID_DEL_SLOT@";
private static final String SEPARATOR = "@";
private static final ConcurrentHashMap<String, Object> SEQUENCES = new ConcurrentHashMap<>();
public static long CID_DEL_TIMEOUT = 24 * 3600 * 1000;
private static long CID_DEL_TIMEOUT = 24 * 3600 * 1000;
private final long clusterId;
public IdMetaStore(PDConfig pdConfig) {

View File

@ -0,0 +1,151 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>hugegraph-pd</artifactId>
<groupId>org.apache.hugegraph</groupId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hg-pd-dist</artifactId>
<properties>
<dist.dir>${project.parent.basedir}/dist</dist.dir>
<shell-executable>bash</shell-executable>
<assembly.dir>${project.basedir}/src/assembly</assembly.dir>
<assembly.descriptor.dir>${assembly.dir}/descriptor</assembly.descriptor.dir>
<assembly.static.dir>${assembly.dir}/static</assembly.static.dir>
<final.name>hugegraph-pd-${project.parent.version}</final.name>
<executable.jar.name>hg-pd-service</executable.jar.name>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-pd-service</artifactId>
<version>${revision}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4</version>
<executions>
<execution>
<id>assembly-hugegraph-pd</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
<configuration>
<attach>false</attach>
<appendAssemblyId>false</appendAssemblyId>
<outputDirectory>${dist.dir}</outputDirectory>
<descriptors>
<descriptor>
${assembly.descriptor.dir}/server-assembly.xml
</descriptor>
</descriptors>
<finalName>${final.name}</finalName>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-clean-plugin</artifactId>
<configuration>
<filesets>
<fileset>
<directory>${dist.dir}</directory>
</fileset>
</filesets>
</configuration>
<executions>
<execution>
<phase>initialize</phase>
<goals>
<goal>clean</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<version>1.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<taskdef resource="net/sf/antcontrib/antcontrib.properties"
classpathref="maven.plugin.classpath" />
<if>
<contains string="${os.name}" substring="Linux" />
<then>
<echo file="${dist.dir}/dist.sh">
tar zcvf \
${dist.dir}/${final.name}.tar.gz \
${final.name} || exit 1
rm -f ${dist.dir}/dist.sh
echo
echo "HugeGraph dist tar.gz available at:
${dist.dir}/${final.name}.tar.gz"
echo
</echo>
<exec executable="${shell-executable}"
dir="${dist.dir}" failonerror="true">
<arg line="./dist.sh" />
</exec>
</then>
<else>
<!-- WARN: keep msg in one line -->
<echo message="We only tar package in Linux env, you could check pom.xml in pd-dist to modify it" />
</else>
</if>
</tasks>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>ant-contrib</groupId>
<artifactId>ant-contrib</artifactId>
<version>1.0b3</version>
<exclusions>
<exclusion>
<groupId>ant</groupId>
<artifactId>ant</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</project>

View File

@ -0,0 +1,57 @@
<!--
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.
-->
<assembly>
<id>distribution</id>
<includeBaseDirectory>false</includeBaseDirectory>
<formats>
<format>dir</format>
</formats>
<fileSets>
<fileSet>
<directory>${assembly.static.dir}/bin</directory>
<outputDirectory>bin</outputDirectory>
<includes>
<include>*</include>
</includes>
<fileMode>755</fileMode>
</fileSet>
<fileSet>
<directory>${assembly.static.dir}/conf</directory>
<outputDirectory>conf</outputDirectory>
<includes>
<include>*</include>
</includes>
</fileSet>
</fileSets>
<dependencySets>
<!-- code jars -->
<dependencySet>
<outputDirectory>/lib</outputDirectory>
<unpack>false</unpack>
<scope>runtime</scope>
<useProjectArtifact>false</useProjectArtifact>
<includes>
<include>org.apache.hugegraph:${executable.jar.name}:jar:*</include>
</includes>
</dependencySet>
</dependencySets>
</assembly>

View File

@ -0,0 +1,169 @@
#!/bin/bash
#
# 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.
#
if [ -z "$GC_OPTION" ];then
GC_OPTION=""
fi
if [ -z "$USER_OPTION" ];then
USER_OPTION=""
fi
if [ -z "$OPEN_TELEMETRY" ];then
OPEN_TELEMETRY="false"
fi
while getopts "g:j:y:" arg; do
case ${arg} in
g) GC_OPTION="$OPTARG" ;;
j) USER_OPTION="$OPTARG" ;;
# Telemetry is used to collect metrics, traces and logs
y) OPEN_TELEMETRY="$OPTARG" ;;
?) echo "USAGE: $0 [-g g1] [-j xxx] [-y true|false]" && exit 1 ;;
esac
done
function abs_path() {
SOURCE="${BASH_SOURCE[0]}"
while [ -h "$SOURCE" ]; do
DIR="$( cd -P "$( dirname "$SOURCE" )" && pwd )"
SOURCE="$(readlink "$SOURCE")"
[[ $SOURCE != /* ]] && SOURCE="$DIR/$SOURCE"
done
echo "$( cd -P "$( dirname "$SOURCE" )" && pwd )"
}
BIN=$(abs_path)
TOP="$(cd "$BIN"/../ && pwd)"
CONF="$TOP/conf"
LIB="$TOP/lib"
PLUGINS="$TOP/plugins"
LOGS="$TOP/logs"
OUTPUT=${LOGS}/hugegraph-pd-stdout.log
GITHUB="https://github.com"
PID_FILE="$BIN/pid"
. "$BIN"/util.sh
ensure_path_writable "$LOGS"
ensure_path_writable "$PLUGINS"
# The maximum and minimum heap memory that service can use
MAX_MEM=$((32 * 1024))
MIN_MEM=$((1 * 512))
EXPECT_JDK_VERSION=11
# Change to $BIN's parent
cd "${TOP}" || exit
# Find Java
if [ "$JAVA_HOME" = "" ]; then
JAVA="java"
else
JAVA="$JAVA_HOME/bin/java"
fi
# check jdk version
JAVA_VERSION=$($JAVA -version 2>&1 | awk 'NR==1{gsub(/"/,""); print $3}' | awk -F'_' '{print $1}')
if [[ $? -ne 0 || $JAVA_VERSION < $EXPECT_JDK_VERSION ]]; then
echo "Please make sure that the JDK is installed and the version >= $EXPECT_JDK_VERSION" >> ${OUTPUT}
exit 1
fi
# Set Java options
if [ "$JAVA_OPTIONS" = "" ]; then
XMX=$(calc_xmx $MIN_MEM $MAX_MEM)
if [ $? -ne 0 ]; then
echo "Failed to start HugeGraphPDServer, requires at least ${MIN_MEM}m free memory" \
>> ${OUTPUT}
exit 1
fi
JAVA_OPTIONS="-Xms${MIN_MEM}m -Xmx${XMX}m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=${LOGS} ${USER_OPTION}"
# Rolling out detailed GC logs
#JAVA_OPTIONS="${JAVA_OPTIONS} -XX:+UseGCLogFileRotation -XX:GCLogFileSize=10M -XX:NumberOfGCLogFiles=3 \
# -Xloggc:./logs/gc.log -XX:+PrintHeapAtGC -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
fi
# Using G1GC as the default garbage collector (Recommended for large memory machines)
case "$GC_OPTION" in
g1)
echo "Using G1GC as the default garbage collector"
JAVA_OPTIONS="${JAVA_OPTIONS} -XX:+UseG1GC -XX:+ParallelRefProcEnabled \
-XX:InitiatingHeapOccupancyPercent=50 -XX:G1RSetUpdatingPauseTimePercent=5"
;;
"") ;;
*)
echo "Unrecognized gc option: '$GC_OPTION', only support 'g1' now" >> ${OUTPUT}
exit 1
esac
if [ "${OPEN_TELEMETRY}" == "true" ]; then
OT_JAR="opentelemetry-javaagent.jar"
OT_JAR_PATH="${PLUGINS}/${OT_JAR}"
if [[ ! -e "${OT_JAR_PATH}" ]]; then
echo "## Downloading ${OT_JAR}..."
download "${PLUGINS}" \
"${GITHUB}/open-telemetry/opentelemetry-java-instrumentation/releases/download/v2.1.0/${OT_JAR}"
if [[ ! -e "${OT_JAR_PATH}" ]]; then
echo "## Error: Failed to download ${OT_JAR}." >>${OUTPUT}
exit 1
fi
fi
# Note: remember update it if we change the jar
expected_md5="e3bcbbe8ed9b6d840fa4c333b36f369f"
actual_md5=$(md5sum "${OT_JAR_PATH}" | awk '{print $1}')
if [[ "${expected_md5}" != "${actual_md5}" ]]; then
echo "## Error: MD5 checksum verification failed for ${OT_JAR_PATH}." >>${OUTPUT}
echo "## Tips: Remove the file and try again." >>${OUTPUT}
exit 1
fi
# Note: check carefully if multi "javeagent" params are set
export JAVA_TOOL_OPTIONS="-javaagent:${PLUGINS}/${OT_JAR}"
export OTEL_TRACES_EXPORTER=otlp
export OTEL_METRICS_EXPORTER=none
export OTEL_LOGS_EXPORTER=none
export OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=grpc
# 127.0.0.1:4317 is the port of otel-collector running in Docker located in
# 'hugegraph-server/hugegraph-dist/docker/example/docker-compose-trace.yaml'.
# Make sure the otel-collector is running before starting HugeGraphPD.
export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://127.0.0.1:4317
export OTEL_RESOURCE_ATTRIBUTES=service.name=pd
fi
#if [ "${JMX_EXPORT_PORT}" != "" ] && [ ${JMX_EXPORT_PORT} -ne 0 ] ; then
# JAVA_OPTIONS="${JAVA_OPTIONS} -javaagent:${LIB}/jmx_prometheus_javaagent-0.16.1.jar=${JMX_EXPORT_PORT}:${CONF}/jmx_exporter.yml"
#fi
if [ $(ps -ef|grep -v grep| grep java|grep -cE ${CONF}) -ne 0 ]; then
echo "HugeGraphPDServer is already running..."
exit 0
fi
echo "Starting HugeGraphPDServer..."
# Turn on security check
exec ${JAVA} -Dname="HugeGraphPD" ${JAVA_OPTIONS} -jar \
-Dspring.config.location=${CONF}/application.yml ${LIB}/hg-pd-service-*.jar >> ${OUTPUT} 2>&1 &
PID="$!"
# Write pid to file
echo "$PID" > "$PID_FILE"
echo "[+pid] $PID"

View File

@ -0,0 +1,48 @@
#!/bin/bash
#
# 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.
#
abs_path() {
SOURCE="${BASH_SOURCE[0]}"
while [ -h "$SOURCE" ]; do
DIR="$( cd -P "$( dirname "$SOURCE" )" && pwd )"
SOURCE="$(readlink "$SOURCE")"
[[ $SOURCE != /* ]] && SOURCE="$DIR/$SOURCE"
done
echo "$( cd -P "$( dirname "$SOURCE" )" && pwd )"
}
BIN=$(abs_path)
TOP="$(cd $BIN/../ && pwd)"
. "$BIN"/util.sh
PID_FILE=$BIN/pid
SERVER_SHUTDOWN_TIMEOUT_S=30
if [ ! -f ${PID_FILE} ]; then
echo "The pid file $PID_FILE doesn't exist"
exit 0
fi
PID=$(cat $PID_FILE)
kill_process_and_wait "HugeGraphPDServer" "$PID" "$SERVER_SHUTDOWN_TIMEOUT_S"
if [ $? -eq 0 ]; then
rm "$PID_FILE"
fi

View File

@ -0,0 +1,373 @@
#!/bin/bash
#
# 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.
#
# TODO: consider reuse it with server-dist module (almost same as it)
function command_available() {
local cmd=$1
if [ $(command -v $cmd >/dev/null 2>&1) ]; then
return 1
else
return 0
fi
}
# read a property from .properties file
function read_property() {
# file path
file_name=$1
# replace "." to "\."
property_name=$(echo $2 | sed 's/\./\\\./g')
cat $file_name | sed -n -e "s/^[ ]*//g;/^#/d;s/^$property_name=//p" | tail -1
}
function write_property() {
local file=$1
local key=$2
local value=$3
local os=$(uname)
case $os in
# Note: in mac os should use sed -i '' "xxx" to replace string,
# otherwise prompt 'command c expects \ followed by text'.
# See http://www.cnblogs.com/greedy-day/p/5952899.html
Darwin) sed -i '' "s!$key=.*!$key=$value!g" "$file" ;;
*) sed -i "s!$key=.*!$key=$value!g" "$file" ;;
esac
}
function parse_yaml() {
local file=$1
local version=$2
local module=$3
cat $file | tr -d '\n {}'| awk -F',+|:' '''{
pre="";
for(i=1; i<=NF; ) {
if(match($i, /version/)) {
pre=$i;
i+=1
} else {
result[pre"-"$i] = $(i+1);
i+=2
}
}
} END {for(e in result) {print e": "result[e]}}''' \
| grep "$version-$module" | awk -F':' '{print $2}' | tr -d ' ' && echo
}
function process_num() {
num=`ps -ef | grep $1 | grep -v grep | wc -l`
return $num
}
function process_id() {
pid=`ps -ef | grep $1 | grep -v grep | awk '{print $2}'`
return $pid
}
# check the port of rest server is occupied
function check_port() {
local port=`echo $1 | awk -F':' '{print $3}'`
if ! command_available "lsof"; then
echo "Required lsof but it is unavailable"
exit 1
fi
lsof -i :$port >/dev/null
if [ $? -eq 0 ]; then
echo "The port $port has already been used"
exit 1
fi
}
function crontab_append() {
local job="$1"
crontab -l | grep -F "$job" >/dev/null 2>&1
if [ $? -eq 0 ]; then
return 1
fi
(crontab -l ; echo "$job") | crontab -
}
function crontab_remove() {
local job="$1"
# check exist before remove
crontab -l | grep -F "$job" >/dev/null 2>&1
if [ $? -eq 1 ]; then
return 0
fi
crontab -l | grep -Fv "$job" | crontab -
# Check exist after remove
crontab -l | grep -F "$job" >/dev/null 2>&1
if [ $? -eq 0 ]; then
return 1
else
return 0
fi
}
# wait_for_startup friendly_name host port timeout_s
function wait_for_startup() {
local pid="$1"
local server_name="$2"
local server_url="$3"
local timeout_s="$4"
local now_s=`date '+%s'`
local stop_s=$(( $now_s + $timeout_s ))
local status
echo -n "Connecting to $server_name ($server_url)"
while [ $now_s -le $stop_s ]; do
echo -n .
process_status "$server_name" "$pid" >/dev/null
if [ $? -eq 1 ]; then
echo "Starting $server_name failed"
return 1
fi
status=`curl -o /dev/null -s -k -w %{http_code} $server_url`
if [[ $status -eq 200 || $status -eq 401 ]]; then
echo "OK"
echo "Started [pid $pid]"
return 0
fi
sleep 2
now_s=`date '+%s'`
done
echo "The operation timed out when attempting to connect to $server_url" >&2
return 1
}
function free_memory() {
local free=""
local os=`uname`
if [ "$os" == "Linux" ]; then
local mem_free=`cat /proc/meminfo | grep -w "MemFree" | awk '{print $2}'`
local mem_buffer=`cat /proc/meminfo | grep -w "Buffers" | awk '{print $2}'`
local mem_cached=`cat /proc/meminfo | grep -w "Cached" | awk '{print $2}'`
if [[ "$mem_free" == "" || "$mem_buffer" == "" || "$mem_cached" == "" ]]; then
echo "Failed to get free memory"
exit 1
fi
free=`expr $mem_free + $mem_buffer + $mem_cached`
free=`expr $free / 1024`
elif [ "$os" == "Darwin" ]; then
local pages_free=`vm_stat | awk '/Pages free/{print $0}' | awk -F'[:.]+' '{print $2}' | tr -d " "`
local pages_inactive=`vm_stat | awk '/Pages inactive/{print $0}' | awk -F'[:.]+' '{print $2}' | tr -d " "`
local pages_available=`expr $pages_free + $pages_inactive`
free=`expr $pages_available \* 4096 / 1024 / 1024`
else
echo "Unsupported operating system $os"
exit 1
fi
echo $free
}
function calc_xmx() {
local min_mem=$1
local max_mem=$2
# Get machine available memory
local free=`free_memory`
local half_free=$[free/2]
local xmx=$min_mem
if [[ "$free" -lt "$min_mem" ]]; then
exit 1
elif [[ "$half_free" -ge "$max_mem" ]]; then
xmx=$max_mem
elif [[ "$half_free" -lt "$min_mem" ]]; then
xmx=$min_mem
else
xmx=$half_free
fi
echo $xmx
}
function remove_with_prompt() {
local path=$1
local tips=""
if [ -d "$path" ]; then
tips="Remove directory '$path' and all sub files [y/n]?"
elif [ -f "$path" ]; then
tips="Remove file '$path' [y/n]?"
else
return 0
fi
read -p "$tips " yn
case $yn in
[Yy]* ) rm -rf "$path";;
* ) ;;
esac
}
function ensure_path_writable() {
local path=$1
# Ensure input path exist
if [ ! -d "${path}" ]; then
mkdir -p ${path}
fi
# Check for write permission
if [ ! -w "${path}" ]; then
echo "No write permission on directory ${path}"
exit 1
fi
}
function get_ip() {
local os=`uname`
local loopback="127.0.0.1"
local ip=""
case $os in
Linux)
if command_available "ifconfig"; then
ip=`ifconfig | grep 'inet addr:' | grep -v "$loopback" | cut -d: -f2 | awk '{ print $1}'`
elif command_available "ip"; then
ip=`ip addr | grep 'state UP' -A2 | tail -n1 | awk '{print $2}' | awk -F"/" '{print $1}'`
else
ip=$loopback
fi
;;
FreeBSD|OpenBSD|Darwin)
if command_available "ifconfig"; then
ip=`ifconfig | grep -E 'inet.[0-9]' | grep -v "$loopback" | awk '{ print $2}'`
else
ip=$loopback
fi
;;
SunOS)
if command_available "ifconfig"; then
ip=`ifconfig -a | grep inet | grep -v "$loopback" | awk '{ print $2} '`
else
ip=$loopback
fi
;;
*) ip=$loopback;;
esac
echo $ip
}
function download() {
local path=$1
local link_url=$2
if command_available "wget"; then
wget --help | grep -q '\--show-progress' && progress_opt="-q --show-progress" || progress_opt=""
wget ${link_url} -P ${path} $progress_opt
elif command_available "curl"; then
curl ${link_url} -o ${path}/${link_url}
else
echo "Required wget or curl but they are unavailable"
exit 1
fi
}
function ensure_package_exist() {
local path=$1
local dir=$2
local tar=$3
local link=$4
if [ ! -d ${path}/${dir} ]; then
if [ ! -f ${path}/${tar} ]; then
echo "Downloading the compressed package '${tar}'"
download ${path} ${link}
if [ $? -ne 0 ]; then
echo "Failed to download, please ensure the network is available and link is valid"
exit 1
fi
echo "[OK] Finished download"
fi
echo "Unzip the compressed package '$tar'"
tar -zxvf ${path}/${tar} -C ${path} >/dev/null 2>&1
if [ $? -ne 0 ]; then
echo "Failed to unzip, please check the compressed package"
exit 1
fi
echo "[OK] Finished unzip"
fi
}
###########################################################################
function wait_for_shutdown() {
local process_name="$1"
local pid="$2"
local timeout_s="$3"
local now_s=`date '+%s'`
local stop_s=$(( $now_s + $timeout_s ))
echo -n "Killing $process_name(pid $pid)" >&2
while [ $now_s -le $stop_s ]; do
echo -n .
process_status "$process_name" "$pid" >/dev/null
if [ $? -eq 1 ]; then
echo "OK"
return 0
fi
sleep 2
now_s=`date '+%s'`
done
echo "$process_name shutdown timeout(exceeded $timeout_s seconds)" >&2
return 1
}
function process_status() {
local process_name="$1"
local pid="$2"
ps -p "$pid"
if [ $? -eq 0 ]; then
echo "$process_name is running with pid $pid"
return 0
else
echo "The process $process_name does not exist"
return 1
fi
}
function kill_process() {
local process_name="$1"
local pid="$2"
if [ -z "$pid" ]; then
echo "The process $pid does not exist"
return 0
fi
case "`uname`" in
CYGWIN*) taskkill /F /PID "$pid" ;;
*) kill "$pid" ;;
esac
}
function kill_process_and_wait() {
local process_name="$1"
local pid="$2"
local timeout_s="$3"
kill_process "$process_name" "$pid"
wait_for_shutdown "$process_name" "$pid" "$timeout_s"
}

View File

@ -0,0 +1,79 @@
#
# 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.
#
spring:
application:
name: hugegraph-pd
management:
metrics:
export:
prometheus:
enabled: true
endpoints:
web:
exposure:
include: "*"
logging:
config: 'file:./conf/log4j2.xml'
# TODO: handle the license file later (PDConfig)
license:
verify-path: ./conf/verify-license.json
license-path: ./conf/hugegraph.license
grpc:
port: 8686
# grpc 的服务地址,部署时需要改为本地实际 IPv4 地址
host: 127.0.0.1
server:
# rest 服务端口号
port: 8620
pd:
# 存储路径
data-path: ./pd_data
# 自动扩容的检查周期,定时检查每个 store 的分区数量,自动进行分区数量平衡
patrol-interval: 1800
# 初始 store 列表,在列表内的 store 自动激活
initial-store-count: 1
# grpc IP:grpc port
initial-store-list: 127.0.0.1:8500
raft:
# 本机 raft 服务地址
address: 127.0.0.1:8610
# pd 集群服务地址
peers-list: 127.0.0.1:8610
store:
# store 下线时间。超过该时间,认为 store 永久不可用,分配副本到其他机器,单位秒
max-down-time: 172800
# 是否开启 store 监控数据存储
monitor_data_enabled: true
# 监控数据的间隔minute (默认), hour, second
# default: 1 min * 1 day = 1440
monitor_data_interval: 1 minute
# 监控数据的保留时间 1 天; day, month, year
monitor_data_retention: 1 day
initial-store-count: 1
partition:
# 默认每个分区副本数
default-shard-count: 1
# 默认每机器最大副本数,初始分区数 = store-max-shard-count * store-number / default-shard-count
store-max-shard-count: 12

View File

@ -0,0 +1,71 @@
#
# 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.
#
spring:
application:
name: hugegraph-pd
logging:
config: file:./conf/log4j2.xml
management:
metrics:
export:
prometheus:
enabled: true
endpoints:
web:
exposure:
include: "*"
grpc:
port: $GRPC_PORT$
# grpc的服务地址,
#注意部署时需要改为本地实际IPv4地址。
host: $GRPC_HOST$
netty-server:
max-inbound-message-size: 100MB
server:
port : $SERVER_PORT$
pd:
# 集群ID区分不同的PD集群
patrol-interval: 2147483647
data-path: $PD_DATA_PATH$
raft:
address: $RAFT_ADDRESS$
# raft集群
peers-list: $RAFT_PEERS_LIST$
# 快照生成时间间隔,单位秒
snapshotInterval: 300
metrics: true
store:
# store心跳超时时间超过该时间认为store临时不可用转移Leader到其他副本,单位秒
keepAlive-timeout: 60
# store下线时间。超过该时间认为store永久不可用分配副本到其他机器单位秒
max-down-time: 1800
partition:
# 默认分区总数
default-total-count: 30
# 默认每个分区副本数
default-shard-count: 3
discovery:
#客户端注册后,无心跳最长次数,超过后,之前的注册信息会被删除
heartbeat-try-count: 3

View File

@ -0,0 +1,135 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<!-- Config will be auto loaded every 60s -->
<configuration status="error" monitorInterval="60">
<properties>
<property name="LOG_PATH">logs</property>
<property name="FILE_NAME">hugegraph-pd</property>
</properties>
<appenders>
<Console name="console" target="SYSTEM_OUT">
<ThresholdFilter level="DEBUG" onMatch="ACCEPT" onMismatch="DENY" />
<PatternLayout pattern="%-d{yyyy-MM-dd HH:mm:ss} [%t] [%p] %c{1.} - %m%n" />
</Console>
<!-- Normal server log config -->
<RollingRandomAccessFile name="file" fileName="${LOG_PATH}/${FILE_NAME}.log"
filePattern="${LOG_PATH}/$${date:yyyy-MM}/${FILE_NAME}-%d{yyyy-MM-dd}-%i.log"
bufferedIO="true" bufferSize="524288" immediateFlush="false">
<ThresholdFilter level="TRACE" onMatch="ACCEPT" onMismatch="DENY" />
<PatternLayout pattern="%-d{yyyy-MM-dd HH:mm:ss} [%t] [%p] %c{1.} - %m%n" />
<!--JsonLayout compact="true" eventEol="true" complete="true" locationInfo="true">
<KeyValuePair key="timestamp" value="$${date:yyyy-MM-dd HH:mm:ss.SSS}"/>
</JsonLayout-->
<!-- Trigger after exceeding 1day or 50MB -->
<Policies>
<SizeBasedTriggeringPolicy size="128MB" />
<TimeBasedTriggeringPolicy interval="1" modulate="true" />
</Policies>
<!-- Keep 5 files per day & auto delete after over 2GB or 100 files -->
<DefaultRolloverStrategy max="16">
<Delete basePath="${LOG_PATH}" maxDepth="2">
<IfFileName glob="*/*.log" />
<!-- Limit log amount & size -->
<IfAny>
<IfAccumulatedFileSize exceeds="2GB" />
<IfAccumulatedFileCount exceeds="100" />
</IfAny>
</Delete>
</DefaultRolloverStrategy>
</RollingRandomAccessFile>
<!-- jraft server log config -->
<RollingRandomAccessFile name="raft_file" fileName="${LOG_PATH}/${FILE_NAME}_raft.log"
filePattern="${LOG_PATH}/$${date:yyyy-MM}/${FILE_NAME}-%d{yyyy-MM-dd}-%i.log"
bufferedIO="true" bufferSize="524288" immediateFlush="false">
<ThresholdFilter level="TRACE" onMatch="ACCEPT" onMismatch="DENY" />
<PatternLayout pattern="%-d{yyyy-MM-dd HH:mm:ss} [%t] [%p] %c{1.} - %m%n" />
<!--JsonLayout compact="true" eventEol="true" complete="true" locationInfo="true">
<KeyValuePair key="timestamp" value="$${date:yyyy-MM-dd HH:mm:ss.SSS}"/>
</JsonLayout-->
<!-- Trigger after exceeding 1day or 50MB -->
<Policies>
<SizeBasedTriggeringPolicy size="128MB" />
<TimeBasedTriggeringPolicy interval="1" modulate="true" />
</Policies>
<!-- Keep 5 files per day & auto Delete after over 2GB or 100 files -->
<DefaultRolloverStrategy max="16">
<Delete basePath="${LOG_PATH}" maxDepth="2">
<IfFileName glob="*/*.log" />
<!-- Limit log amount & size -->
<IfAny>
<IfAccumulatedFileSize exceeds="2GB" />
<IfAccumulatedFileCount exceeds="100" />
</IfAny>
</Delete>
</DefaultRolloverStrategy>
</RollingRandomAccessFile>
<!-- Separate & compress audit log, buffer size is 512KB -->
<RollingRandomAccessFile name="audit" fileName="${LOG_PATH}/audit-${FILE_NAME}.log"
filePattern="${LOG_PATH}/$${date:yyyy-MM}/audit-${FILE_NAME}-%d{yyyy-MM-dd-HH}-%i.gz"
bufferedIO="true" bufferSize="524288" immediateFlush="false">
<ThresholdFilter level="TRACE" onMatch="ACCEPT" onMismatch="DENY" />
<!-- Use a simple format for audit log to speed up -->
<!-- PatternLayout pattern="%-d{yyyy-MM-dd HH:mm:ss} - %m%n"/ -->
<JsonLayout compact="true" eventEol="true" locationInfo="true">
<KeyValuePair key="timestamp" value="$${date:yyyy-MM-dd HH:mm:ss.SSS}" />
</JsonLayout>
<!-- Trigger after exceeding 1hour or 500MB -->
<Policies>
<SizeBasedTriggeringPolicy size="512MB" />
<TimeBasedTriggeringPolicy interval="1" modulate="true" />
</Policies>
<!-- Keep 2 files per hour & auto delete [after 60 days] or [over 5GB or 500 files] -->
<DefaultRolloverStrategy max="16">
<Delete basePath="${LOG_PATH}" maxDepth="2">
<IfFileName glob="*/*.gz" />
<IfLastModified age="60d" />
<IfAny>
<IfAccumulatedFileSize exceeds="5GB" />
<IfAccumulatedFileCount exceeds="500" />
</IfAny>
</Delete>
</DefaultRolloverStrategy>
</RollingRandomAccessFile>
</appenders>
<loggers>
<root level="INFO">
<appender-ref ref="file" />
</root>
<logger name="com.alipay.sofa" level="INFO" additivity="false">
<appender-ref ref="raft_file" />
</logger>
<logger name="io.netty" level="INFO" additivity="false">
<appender-ref ref="file" />
</logger>
<logger name="org.apache.commons" level="INFO" additivity="false">
<appender-ref ref="file" />
</logger>
<!-- Use mixed async way to output logs -->
<logger name="org.apache.hugegraph" level="INFO" additivity="false">
<appender-ref ref="file" />
<appender-ref ref="console" />
</logger>
</loggers>
</configuration>

View File

@ -0,0 +1,6 @@
{
"subject": "hugegraph-license",
"public_alias": "publiccert",
"store_ticket": "803b6cc3-d144-47e8-948f-ec8b39c8881e",
"publickey_path": "/public-certs.store"
}

View File

@ -1347,6 +1347,7 @@ public class PDService extends PDGrpc.PDImplBase implements ServiceGrpc, RaftSta
observer.onCompleted();
}
// TODO: keep it now & clean it later
@Override
public void putLicense(PutLicenseRequest request,
StreamObserver<PutLicenseResponse> responseObserver) {
@ -1366,12 +1367,12 @@ public class PDService extends PDGrpc.PDImplBase implements ServiceGrpc, RaftSta
}
FileUtils.writeByteArrayToFile(licenseFile, content, false);
} catch (Exception e) {
log.error("putLicense with error: {}", e);
log.error("putLicense with error:", e);
if (moved) {
try {
FileUtils.moveFile(bakFile, licenseFile);
} catch (IOException ex) {
log.error("failed to restore the license file.{}", ex);
log.error("failed to restore the license file:", ex);
}
}
Pdpb.ResponseHeader header =

View File

@ -94,6 +94,11 @@
<version>${log4j2.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-pd-grpc</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-pd-common</artifactId>
@ -114,6 +119,12 @@
<artifactId>hg-pd-service</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-pd-dist</artifactId>
<version>${revision}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
@ -171,20 +182,13 @@
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-junit4</artifactId>
<version>${powermock.version}</version>
<version>2.0.0-RC.3</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito2</artifactId>
<version>${powermock.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.14.0</version>
<version>2.0.0-RC.3</version>
<scope>compile</scope>
</dependency>
</dependencies>
@ -196,6 +200,30 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>2.20</version>
<executions>
<execution>
<id>pd-client-test</id>
<configuration>
<testSourceDirectory>${basedir}/src/main/java/
</testSourceDirectory>
<testClassesDirectory>${basedir}/target/classes/
</testClassesDirectory>
<includes>
<include>**/PDClientSuiteTest.java</include>
</includes>
</configuration>
</execution>
<execution>
<id>pd-core-test</id>
<configuration>
<testSourceDirectory>${basedir}/src/main/java/
</testSourceDirectory>
<testClassesDirectory>${basedir}/target/classes/
</testClassesDirectory>
<includes>
<include>**/PDCoreSuiteTest.java</include>
</includes>
</configuration>
</execution>
<execution>
<id>pd-common-test</id>
<configuration>
@ -209,14 +237,14 @@
</configuration>
</execution>
<execution>
<id>pd-client-test</id>
<id>pd-rest-test</id>
<configuration>
<testSourceDirectory>${basedir}/src/main/java/
</testSourceDirectory>
<testClassesDirectory>${basedir}/target/classes/
</testClassesDirectory>
<includes>
<include>**/PDClientSuiteTest.java</include>
<include>**/PDRestSuiteTest.java</include>
</includes>
</configuration>
</execution>

View File

@ -0,0 +1,41 @@
/*
* 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.hugegraph.pd.client;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.runner.RunWith;
import org.mockito.runners.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class BaseClientTest {
protected static PDClient pdClient;
@BeforeClass
public static void beforeClass() {
PDConfig config = PDConfig.of("localhost:8686");
config.setEnableCache(true);
pdClient = PDClient.create(config);
}
@After
public void teardown() {
// pass
}
}

View File

@ -0,0 +1,58 @@
/*
* 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.hugegraph.pd.client;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.hugegraph.pd.common.Useless;
import com.alipay.sofa.jraft.CliService;
import com.alipay.sofa.jraft.RaftServiceFactory;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.entity.PeerId;
import com.alipay.sofa.jraft.option.CliOptions;
@Useless("used for development")
public class ChangingLeader {
private static final CliService cliService =
RaftServiceFactory.createAndInitCliService(new CliOptions());
public static void main(String[] args) {
var conf = new Configuration();
conf.addPeer(PeerId.parsePeer("127.0.0.1:8610"));
conf.addPeer(PeerId.parsePeer("127.0.0.1:8611"));
conf.addPeer(PeerId.parsePeer("127.0.0.1:8612"));
CountDownLatch latch = new CountDownLatch(100);
Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
Status status = cliService.transferLeader("pd_raft", conf, PeerId.ANY_PEER);
System.out.println("trigger change leader status: " + status);
latch.countDown();
}, 1, 3, TimeUnit.SECONDS);
try {
latch.await();
} catch (Exception e) {
System.out.println(e);
}
}
}

View File

@ -0,0 +1,118 @@
/*
* 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.hugegraph.pd.client;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import java.util.function.Consumer;
import org.apache.commons.lang3.StringUtils;
import org.apache.hugegraph.pd.grpc.kv.KResponse;
import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
import org.apache.hugegraph.pd.grpc.kv.WatchKv;
import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
import org.apache.hugegraph.pd.grpc.kv.WatchState;
import org.apache.hugegraph.pd.grpc.kv.WatchType;
import org.junit.Before;
import org.junit.Test;
import io.grpc.stub.AbstractBlockingStub;
import io.grpc.stub.AbstractStub;
public class KvClientTest extends BaseClientTest {
String key = "key";
String value = "value";
private KvClient<WatchResponse> client;
@Before
public void setUp() {
this.client = new KvClient<>(PDConfig.of("localhost:8686"));
}
@Test
public void testCreateStub() {
// Setup
// Run the test
try {
final AbstractStub result = this.client.createStub();
} catch (Exception e) {
}
// Verify the results
}
@Test
public void testCreateBlockingStub() {
// Setup
// Run the test
try {
final AbstractBlockingStub result = this.client.createBlockingStub();
} catch (Exception e) {
}
}
@Test
public void testPutAndGet() throws Exception {
// Run the test
try {
this.client.put(this.key, this.value);
// Run the test
KResponse result = this.client.get(this.key);
// Verify the results
assertThat(result.getValue()).isEqualTo(this.value);
this.client.delete(this.key);
result = this.client.get(this.key);
assertThat(StringUtils.isEmpty(result.getValue()));
this.client.deletePrefix(this.key);
this.client.put(this.key + "1", this.value);
this.client.put(this.key + "2", this.value);
ScanPrefixResponse response = this.client.scanPrefix(this.key);
assertThat(response.getKvsMap().size() == 2);
this.client.putTTL(this.key + "3", this.value, 1000);
this.client.keepTTLAlive(this.key + "3");
final Consumer<WatchResponse> mockConsumer = mock(Consumer.class);
// Run the test
this.client.listen(this.key + "3", mockConsumer);
this.client.listenPrefix(this.key + "4", mockConsumer);
WatchResponse r = WatchResponse.newBuilder().addEvents(
WatchEvent.newBuilder().setCurrent(
WatchKv.newBuilder().setKey(this.key).setValue("value")
.build()).setType(WatchType.Put).build())
.setClientId(0L)
.setState(WatchState.Starting)
.build();
this.client.getWatchList(r);
this.client.getWatchMap(r);
this.client.lock(this.key, 3000L);
this.client.isLocked(this.key);
this.client.unlock(this.key);
this.client.lock(this.key, 3000L);
this.client.keepAlive(this.key);
this.client.close();
} catch (Exception e) {
}
}
}

View File

@ -0,0 +1,34 @@
/*
* 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.hugegraph.pd.client;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
import lombok.extern.slf4j.Slf4j;
@RunWith(Suite.class)
@Suite.SuiteClasses({
PDClientTest.class,
KvClientTest.class,
StoreRegisterTest.class,
})
@Slf4j
public class PDClientSuiteTest {
}

View File

@ -0,0 +1,413 @@
/*
* 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.hugegraph.pd.client;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.grpc.MetaTask;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.pd.grpc.Pdpb;
import org.junit.Test;
import org.mockito.Mockito;
// TODO: Exceptions should be thrown rather than silenced.
public class PDClientTest extends BaseClientTest {
@Test
public void testDbCompaction() {
try {
pdClient.dbCompaction("");
pdClient.dbCompaction();
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testRegisterStore() {
Metapb.Store store = Metapb.Store.newBuilder().build();
try {
pdClient.registerStore(store);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testSetGraph() {
Metapb.Graph graph = Metapb.Graph.newBuilder().setGraphName("test").build();
try {
pdClient.setGraph(graph);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetGraph() {
try {
pdClient.getGraph("test");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetStore() {
try {
pdClient.getStore(0L);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testUpdateStore() {
Metapb.Store store = Metapb.Store.newBuilder().build();
try {
pdClient.updateStore(store);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetActiveStores() {
try {
pdClient.getActiveStores("test");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetAllStores() {
try {
pdClient.getAllStores("test");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testStoreHeartbeat() {
Metapb.StoreStats stats = Metapb.StoreStats.newBuilder().build();
try {
pdClient.storeHeartbeat(stats);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testKeyToCode() {
pdClient.keyToCode("test", "test".getBytes(StandardCharsets.UTF_8));
}
@Test
public void testScanPartitions() {
try {
pdClient.scanPartitions("test", "1".getBytes(StandardCharsets.UTF_8),
"9".getBytes(StandardCharsets.UTF_8));
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetPartitionsByStore() {
try {
pdClient.getPartitionsByStore(0L);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testQueryPartitions() {
try {
pdClient.queryPartitions(0L, 0);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetPartitions() {
try {
pdClient.getPartitions(0L, "test");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testUpdatePartitionLeader() {
pdClient.updatePartitionLeader("aaa", 0, 0L);
}
@Test
public void testInvalidPartitionCache() {
pdClient.invalidPartitionCache();
}
@Test
public void testInvalidStoreCache() {
pdClient.invalidStoreCache(0L);
}
@Test
public void testUpdatePartitionCache() {
Metapb.Partition partition = Metapb.Partition.newBuilder().build();
Metapb.Shard leader = Metapb.Shard.newBuilder().build();
pdClient.updatePartitionCache(partition, leader);
}
@Test
public void testGetIdByKey() {
try {
pdClient.getIdByKey("test", 1);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testResetIdByKey() {
try {
pdClient.resetIdByKey("test");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetGetLeader() {
try {
pdClient.getLeader();
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetMembers() {
try {
pdClient.getMembers();
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetClusterStats() {
try {
pdClient.getClusterStats();
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testAddEventListener() {
PDClient.PDEventListener listener = Mockito.mock(PDClient.PDEventListener.class);
pdClient.addEventListener(listener);
}
@Test
public void testGetWatchClient() {
pdClient.getWatchClient();
}
@Test
public void testGetPulseClient() {
// pdClient.getPulseClient();
}
@Test
public void testGetStoreStatus() {
try {
pdClient.getStoreStatus(true);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetPartition() {
try {
pdClient.getPartition("test", "test".getBytes(StandardCharsets.UTF_8));
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testSetGraphSpace() {
try {
pdClient.setGraphSpace("test", 1L);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetGraphSpace() {
try {
pdClient.getGraphSpace("test");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testSetPDConfig() {
try {
pdClient.setPDConfig(0, "", 0, 0L);
} catch (PDException e) {
e.printStackTrace();
}
Metapb.PDConfig pdConfig = Metapb.PDConfig.newBuilder().build();
try {
pdClient.setPDConfig(pdConfig);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testGetPDConfig() {
try {
pdClient.getPDConfig(0L);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testChangePeerList() {
try {
pdClient.changePeerList("");
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testSplitData() {
try {
Metapb.PDConfig config = pdClient.getPDConfig();
pdClient.setPDConfig(config.toBuilder().setMaxShardsPerStore(12).build());
System.out.println(pdClient.getPDConfig());
// 开始分区分裂
pdClient.splitData();
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testBalancePartition() {
try {
pdClient.balancePartition();
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testMovePartition() {
Pdpb.OperationMode mode = Pdpb.OperationMode.Auto;
List<Pdpb.MovePartitionParam> params = new ArrayList<>(1);
try {
pdClient.movePartition(mode, params);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testReportTask() {
MetaTask.Task task = MetaTask.Task.newBuilder().build();
try {
pdClient.reportTask(task);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testBalanceLeaders() {
try {
pdClient.balanceLeaders();
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testDelStore() {
try {
pdClient.delStore(0L);
} catch (PDException e) {
e.printStackTrace();
}
}
// @Test
// public void testgetQuota() {
// try {
// pdClient.getQuota();
// } catch (PDException e) {
// e.printStackTrace();
// }
// }
@Test
public void testUpdatePartition() {
List<Metapb.Partition> partitions = new ArrayList<>(1);
try {
pdClient.updatePartition(partitions);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testDelPartition() {
try {
pdClient.delPartition("test", 0);
} catch (PDException e) {
e.printStackTrace();
}
}
@Test
public void testdelGraph() {
try {
pdClient.delGraph("test");
} catch (PDException e) {
e.printStackTrace();
}
}
}

View File

@ -0,0 +1,104 @@
/*
* 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.hugegraph.pd.client;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.hugegraph.pd.common.Useless;
import org.apache.hugegraph.pd.grpc.pulse.PartitionHeartbeatRequest;
import org.apache.hugegraph.pd.pulse.PulseServerNotice;
import org.junit.BeforeClass;
import org.junit.Test;
@Useless("used for development")
public class PDPulseTest {
private static PDClient pdClient;
@BeforeClass
public static void beforeClass() throws Exception {
PDConfig pdConfig = PDConfig.of("localhost:8686");
pdConfig.setEnableCache(true);
pdClient = PDClient.create(pdConfig);
pdClient.getLeader();
}
@Test
public void listen() {
PDPulse pulse = new PDPulseImpl(pdClient.getLeaderIp());
CountDownLatch latch = new CountDownLatch(60);
PDPulse.Notifier<PartitionHeartbeatRequest.Builder> notifier1 =
pulse.connectPartition(new PulseListener<>(latch, "listener1"));
PDPulse.Notifier<PartitionHeartbeatRequest.Builder> notifier2 =
pulse.connectPartition(new PulseListener<>(latch, "listener2"));
PDPulse.Notifier<PartitionHeartbeatRequest.Builder> notifier3 =
pulse.connectPartition(new PulseListener<>(latch, "listener3"));
try {
latch.await(120, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
PartitionHeartbeatRequest.Builder builder = PartitionHeartbeatRequest.newBuilder();
notifier1.notifyServer(builder);
notifier2.notifyServer(builder);
notifier3.notifyServer(builder);
notifier1.close();
notifier2.close();
notifier3.close();
}
private static class PulseListener<T> implements PDPulse.Listener<T> {
private final String listenerName;
private final CountDownLatch latch;
private PulseListener(CountDownLatch latch, String listenerName) {
this.latch = latch;
this.listenerName = listenerName;
}
@Override
public void onNext(T response) {
System.out.println(this.listenerName + " ---> res: " + response);
this.latch.countDown();
}
@Override
public void onNotice(PulseServerNotice<T> notice) {
System.out.println(this.listenerName + " ---> res: " + notice.getContent());
notice.ack();
this.latch.countDown();
}
@Override
public void onError(Throwable throwable) {
System.out.println(this.listenerName + " error: " + throwable.toString());
}
@Override
public void onCompleted() {
System.out.println(this.listenerName + " is completed");
}
}
}

View File

@ -0,0 +1,82 @@
/*
* 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.hugegraph.pd.client;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.hugegraph.pd.common.Useless;
import org.junit.BeforeClass;
import org.junit.Test;
@Useless("used for development")
public class PDWatchTest {
private static PDClient pdClient;
@BeforeClass
public static void beforeClass() {
pdClient = PDClient.create(PDConfig.of("localhost:8686"));
}
@Test
public void watch() {
PDWatch watch = pdClient.getWatchClient();
CountDownLatch latch = new CountDownLatch(10);
PDWatch.Watcher watcher1 = watch.watchPartition(new WatchListener<>(latch, "watcher1"));
PDWatch.Watcher watcher2 = watch.watchPartition(new WatchListener<>(latch, "watcher2"));
PDWatch.Watcher watcher3 = watch.watchPartition(new WatchListener<>(latch, "watcher3"));
try {
latch.await(15, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
watcher1.close();
watcher2.close();
watcher3.close();
}
private class WatchListener<T> implements PDWatch.Listener<T> {
private final String watcherName;
CountDownLatch latch;
private WatchListener(CountDownLatch latch, String watcherName) {
this.latch = latch;
this.watcherName = watcherName;
}
@Override
public void onNext(T response) {
System.out.println(this.watcherName + " res: " + response);
this.latch.countDown();
}
@Override
public void onError(Throwable throwable) {
System.out.println(this.watcherName + " error: " + throwable.toString());
}
@Override
public void onCompleted() {
System.out.println(this.watcherName + " is completed");
}
}
}

View File

@ -0,0 +1,127 @@
/*
* 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.hugegraph.pd.client;
import java.nio.charset.StandardCharsets;
import java.util.List;
import org.apache.hugegraph.pd.common.KVPair;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.pd.grpc.pulse.PartitionHeartbeatRequest;
import org.apache.hugegraph.pd.grpc.pulse.PulseResponse;
import org.apache.hugegraph.pd.pulse.PulseServerNotice;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
public class StoreRegisterTest {
private static PDClient pdClient;
private final String storeAddr = "localhost";
private final String graphName = "default/hugegraph/g";
private long storeId = 0;
@BeforeClass
public static void beforeClass() {
PDConfig config = PDConfig.of("localhost:8686");
config.setEnableCache(true);
pdClient = PDClient.create(config);
}
@Test
public void testRegisterStore() throws PDException {
Metapb.Store store = Metapb.Store.newBuilder().setAddress(storeAddr).build();
storeId = pdClient.registerStore(store);
Assert.assertTrue("RegisterStore store_id = " + storeId, storeId != 0);
}
@Test
public void testGetStore() throws PDException {
testRegisterStore();
Metapb.Store store = pdClient.getStore(storeId);
Assert.assertEquals(storeAddr, store.getAddress());
System.out.println(store);
}
@Ignore // no active store
@Test
public void testGetActiveStores() throws PDException {
testRegisterStore();
List<Metapb.Store> stores = pdClient.getActiveStores(graphName);
stores.forEach((e) -> {
System.out.println("-------------------------------------");
System.out.println(e);
});
}
@Ignore // no active store
@Test
public void testStoreHeartbeat() throws PDException {
testRegisterStore();
Metapb.StoreStats stats = Metapb.StoreStats.newBuilder().setStoreId(storeId).build();
pdClient.storeHeartbeat(stats);
List<Metapb.Store> stores = pdClient.getActiveStores(graphName);
boolean exist = false;
for (Metapb.Store store : stores) {
if (store.getId() == storeId) {
exist = true;
break;
}
}
Assert.assertTrue(exist);
}
@Ignore // no active store
@Test
public void testPartitionHeartbeat() throws PDException {
testRegisterStore();
PDPulse pdPulse = new PDPulseImpl(pdClient.getLeaderIp());
PDPulse.Notifier<PartitionHeartbeatRequest.Builder> notifier = pdPulse.connectPartition(
new PDPulse.Listener<>() {
@Override
public void onNext(PulseResponse response) {
}
@Override
public void onNotice(PulseServerNotice<PulseResponse> notice) {
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onCompleted() {
}
});
KVPair<Metapb.Partition, Metapb.Shard> partShard =
pdClient.getPartition("test", "1".getBytes(StandardCharsets.UTF_8));
notifier.notifyServer(PartitionHeartbeatRequest.newBuilder().setStates(
Metapb.PartitionStats.newBuilder().addGraphName("test")
.setId(partShard.getKey().getId())
.setLeader(Metapb.Shard.newBuilder().setStoreId(1).build())));
}
}

View File

@ -29,7 +29,6 @@ import lombok.extern.slf4j.Slf4j;
HgAssertTest.class,
KVPairTest.class,
})
@Slf4j
public class CommonSuiteTest {

View File

@ -384,5 +384,4 @@ public class PartitionCacheTest {
assertEquals(this.cache.getLeaderShard(shardGroup.getId()), leader);
}
}

View File

@ -25,7 +25,7 @@ import org.junit.Test;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class PartitionUtilsTest extends BaseCommonTest {
public class PartitionUtilsTest {
@Test
public void testCalcHashcode() {
@ -34,7 +34,7 @@ public class PartitionUtilsTest extends BaseCommonTest {
Assert.assertEquals(code, 31912L);
}
// @Test
@Test
public void testHashCode() {
int partCount = 10;
int partSize = PartitionUtils.MAX_VALUE / partCount + 1;
@ -42,11 +42,8 @@ public class PartitionUtilsTest extends BaseCommonTest {
for (int i = 0; i < 10000; i++) {
String s = String.format("BATCH-GET-UNIT-%02d", i);
int c = PartitionUtils.calcHashcode(s.getBytes(StandardCharsets.UTF_8));
counter[c / partSize]++;
}
for (int i = 0; i < counter.length; i++) {
System.out.println(i + " " + counter[i]);
}

View File

@ -0,0 +1,105 @@
/*
* 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.hugegraph.pd.core;
import java.util.List;
import org.apache.hugegraph.pd.ConfigService;
import org.apache.hugegraph.pd.IdService;
import org.apache.hugegraph.pd.config.PDConfig;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class ConfigServiceTest extends PDCoreTestBase {
private ConfigService service;
@Before
public void setUp() {
this.service = new ConfigService(getPdConfig());
}
@Test
public void testGetPDConfig() {
// Setup
try {
final Metapb.PDConfig config = Metapb.PDConfig.newBuilder()
.setVersion(0L)
.setPartitionCount(0)
.setShardCount(55)
.setMaxShardsPerStore(0)
.setTimestamp(0L).build();
this.service.setPDConfig(config);
// Run the test
Metapb.PDConfig result = this.service.getPDConfig(0L);
// Verify the results
Assert.assertEquals(55, result.getShardCount());
result = this.service.getPDConfig();
Assert.assertEquals(55, result.getShardCount());
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testGetGraphSpace() throws Exception {
// Setup
Metapb.GraphSpace space = Metapb.GraphSpace.newBuilder()
.setName("gs1")
.setTimestamp(0L).build();
this.service.setGraphSpace(space);
// Run the test
final List<Metapb.GraphSpace> result = this.service.getGraphSpace("gs1");
// Verify the results
Assert.assertEquals(1, result.size());
Assert.assertEquals(space.getName(), result.get(0).getName());
}
@Test
public void testUpdatePDConfig() {
try {
final Metapb.PDConfig mConfig = Metapb.PDConfig.newBuilder()
.setVersion(0L)
.setPartitionCount(0)
.setShardCount(0)
.setMaxShardsPerStore(0)
.setTimestamp(0L)
.build();
final PDConfig expectedResult = new PDConfig();
expectedResult.setConfigService(new ConfigService(new PDConfig()));
expectedResult.setIdService(new IdService(new PDConfig()));
expectedResult.setClusterId(0L);
expectedResult.setPatrolInterval(0L);
expectedResult.setDataPath("dataPath");
expectedResult.setMinStoreCount(0);
expectedResult.setInitialStoreList("initialStoreList");
expectedResult.setHost("host");
expectedResult.setVerifyPath("verifyPath");
expectedResult.setLicensePath("licensePath");
this.service.updatePDConfig(mConfig);
} catch (Exception e) {
e.printStackTrace();
}
}
}

View File

@ -0,0 +1,97 @@
/*
* 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.hugegraph.pd.core;
import org.apache.hugegraph.pd.IdService;
import org.apache.hugegraph.pd.config.PDConfig;
import org.junit.Assert;
import org.junit.Test;
public class IdServiceTest extends PDCoreTestBase {
@Test
public void testCid() {
try {
PDConfig pdConfig = getPdConfig();
int max = 0x2000;
IdService idService = new IdService(pdConfig);
for (int i = 0; i < max; i++) {
idService.getCId("test", max);
}
idService.delCId("test", 1);
idService.delCId("test", 0x10);
idService.delCId("test", 0x100);
idService.delCId("test", 0x1000);
Assert.assertEquals(1, idService.getCId("test", max));
Assert.assertEquals(0x10, idService.getCId("test", max));
Assert.assertEquals(0x100, idService.getCId("test", max));
Assert.assertEquals(0x1000, idService.getCId("test", max));
Assert.assertEquals(-1, idService.getCId("test", max));
idService.delCId("test", 1);
idService.delCId("test", 0x10);
idService.delCId("test", 0x100);
idService.delCId("test", 0x1000);
long cid1 = idService.getCId("test", "name", max);
idService.delCIdDelay("test", "name", cid1);
long cid2 = idService.getCId("test", "name", max);
Assert.assertEquals(cid1, cid2);
idService.delCIdDelay("test", "name", cid2);
Thread.sleep(5000);
long cid3 = idService.getCId("test", "name", max);
} catch (Exception e) {
e.printStackTrace();
}
// MetadataFactory.closeStore();
}
@Test
public void testId() {
try {
PDConfig pdConfig = getPdConfig();
IdService idService = new IdService(pdConfig);
long first = idService.getId("abc", 100);
Assert.assertEquals(first, 0L);
long second = idService.getId("abc", 100);
Assert.assertEquals(second, 100L);
idService.resetId("abc");
first = idService.getId("abc", 100);
Assert.assertEquals(first, 0L);
} catch (Exception e) {
e.printStackTrace();
}
// MetadataFactory.closeStore();
}
@Test
public void testMember() {
try {
PDConfig pdConfig = getPdConfig();
IdService idService = new IdService(pdConfig);
idService.setPdConfig(pdConfig);
PDConfig config = idService.getPdConfig();
config.getHost();
} catch (Exception e) {
e.printStackTrace();
}
// MetadataFactory.closeStore();
}
}

View File

@ -0,0 +1,61 @@
/*
* 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.hugegraph.pd.core;
import org.apache.hugegraph.pd.KvService;
import org.apache.hugegraph.pd.config.PDConfig;
import org.junit.Assert;
import org.junit.Test;
public class KvServiceTest extends PDCoreTestBase {
@Test
public void testKv() {
try {
PDConfig pdConfig = getPdConfig();
KvService service = new KvService(pdConfig);
String key = "kvTest";
String kvTest = service.get(key);
Assert.assertEquals(kvTest, "");
service.put(key, "kvTestValue");
kvTest = service.get(key);
Assert.assertEquals(kvTest, "kvTestValue");
service.scanWithPrefix(key);
service.delete(key);
service.put(key, "kvTestValue");
service.deleteWithPrefix(key);
service.put(key, "kvTestValue", 1000L);
service.keepAlive(key);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testMember() {
try {
PDConfig pdConfig = getPdConfig();
KvService service = new KvService(pdConfig);
service.setPdConfig(pdConfig);
PDConfig config = service.getPdConfig();
// TODO
} catch (Exception e) {
e.printStackTrace();
}
}
}

View File

@ -0,0 +1,50 @@
/*
* 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.hugegraph.pd.core;
import java.util.List;
import org.apache.hugegraph.pd.LogService;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import com.google.protobuf.Any;
public class LogServiceTest extends PDCoreTestBase {
private LogService logServiceUnderTest;
@Before
public void setUp() {
this.logServiceUnderTest = new LogService(getPdConfig());
}
@Test
public void testGetLog() throws Exception {
this.logServiceUnderTest.insertLog("action", "message", Any.newBuilder().build());
// Run the test
final List<Metapb.LogRecord> result =
this.logServiceUnderTest.getLog("action", 0L, System.currentTimeMillis());
// Verify the results
Assert.assertEquals(result.size(), 1);
}
}

View File

@ -0,0 +1,43 @@
/*
* 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.hugegraph.pd.core;
import org.apache.hugegraph.pd.core.meta.MetadataKeyHelperTest;
import org.apache.hugegraph.pd.core.store.HgKVStoreImplTest;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
import lombok.extern.slf4j.Slf4j;
@RunWith(Suite.class)
@Suite.SuiteClasses({
MetadataKeyHelperTest.class,
HgKVStoreImplTest.class,
ConfigServiceTest.class,
IdServiceTest.class,
KvServiceTest.class,
LogServiceTest.class,
PartitionServiceTest.class,
StoreMonitorDataServiceTest.class,
StoreServiceTest.class,
TaskScheduleServiceTest.class
})
@Slf4j
public class PDCoreSuiteTest {
}

View File

@ -0,0 +1,219 @@
/*
* 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.hugegraph.pd.core;
import java.io.File;
import org.apache.hugegraph.pd.ConfigService;
import org.apache.hugegraph.pd.IdService;
import org.apache.hugegraph.pd.PartitionInstructionListener;
import org.apache.hugegraph.pd.PartitionService;
import org.apache.hugegraph.pd.PartitionStatusListener;
import org.apache.hugegraph.pd.StoreMonitorDataService;
import org.apache.hugegraph.pd.StoreNodeService;
import org.apache.hugegraph.pd.StoreStatusListener;
import org.apache.hugegraph.pd.TaskScheduleService;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.config.PDConfig;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.pd.grpc.pulse.ChangeShard;
import org.apache.hugegraph.pd.grpc.pulse.CleanPartition;
import org.apache.hugegraph.pd.grpc.pulse.DbCompaction;
import org.apache.hugegraph.pd.grpc.pulse.MovePartition;
import org.apache.hugegraph.pd.grpc.pulse.PartitionKeyRange;
import org.apache.hugegraph.pd.grpc.pulse.SplitPartition;
import org.apache.hugegraph.pd.grpc.pulse.TransferLeader;
import org.apache.hugegraph.pd.raft.RaftEngine;
import org.junit.AfterClass;
import org.junit.BeforeClass;
public class PDCoreTestBase {
private static final String DATA_PATH = "/tmp/pd_data";
private static PDConfig pdConfig;
private static StoreNodeService storeNodeService;
private static PartitionService partitionService;
private static TaskScheduleService taskService;
private static StoreMonitorDataService storeMonitorDataService;
@BeforeClass
public static void initService() throws PDException {
deleteDir(new File(DATA_PATH));
PDConfig config = new PDConfig();
config.setDataPath(DATA_PATH);
config.setMinStoreCount(3);
config.setInitialStoreList("127.0.0.1:8501");
config.setHost("127.0.0.1");
config.setVerifyPath("");
config.setLicensePath("");
PDConfig.Raft raft = new PDConfig().new Raft();
raft.setAddress("127.0.0.1:8601");
raft.setPeersList("127.0.0.1:8601");
raft.setDataPath(DATA_PATH);
raft.setHost("127.0.0.1");
raft.setGrpcPort(8688);
raft.setPort(8621);
config.setRaft(raft);
config.setStore(new PDConfig().new Store());
config.setPartition(new PDConfig().new Partition() {{
setShardCount(1);
setTotalCount(12);
setMaxShardsPerStore(12);
}});
config.setDiscovery(new PDConfig().new Discovery());
pdConfig = config;
var configService = new ConfigService(pdConfig);
configService.loadConfig();
var engine = RaftEngine.getInstance();
engine.addStateListener(configService);
engine.init(pdConfig.getRaft());
engine.waitingForLeader(5000);
storeNodeService = new StoreNodeService(pdConfig);
partitionService = new PartitionService(pdConfig, storeNodeService);
taskService = new TaskScheduleService(pdConfig, storeNodeService, partitionService);
var idService = new IdService(pdConfig);
storeMonitorDataService = new StoreMonitorDataService(pdConfig);
RaftEngine.getInstance().addStateListener(partitionService);
pdConfig.setIdService(idService);
storeNodeService.init(partitionService);
partitionService.init();
partitionService.addInstructionListener(new PartitionInstructionListener() {
@Override
public void changeShard(Metapb.Partition partition, ChangeShard changeShard) throws
PDException {
}
@Override
public void transferLeader(Metapb.Partition partition,
TransferLeader transferLeader) throws PDException {
}
@Override
public void splitPartition(Metapb.Partition partition,
SplitPartition splitPartition) throws PDException {
}
@Override
public void dbCompaction(Metapb.Partition partition, DbCompaction dbCompaction) throws
PDException {
}
@Override
public void movePartition(Metapb.Partition partition,
MovePartition movePartition) throws PDException {
}
@Override
public void cleanPartition(Metapb.Partition partition,
CleanPartition cleanPartition) throws PDException {
}
@Override
public void changePartitionKeyRange(Metapb.Partition partition,
PartitionKeyRange partitionKeyRange)
throws PDException {
}
});
partitionService.addStatusListener(new PartitionStatusListener() {
@Override
public void onPartitionChanged(Metapb.Partition partition,
Metapb.Partition newPartition) {
}
@Override
public void onPartitionRemoved(Metapb.Partition partition) {
}
});
storeNodeService.addStatusListener(new StoreStatusListener() {
@Override
public void onStoreStatusChanged(Metapb.Store store, Metapb.StoreState old,
Metapb.StoreState status) {
}
@Override
public void onGraphChange(Metapb.Graph graph, Metapb.GraphState stateOld,
Metapb.GraphState stateNew) {
}
@Override
public void onStoreRaftChanged(Metapb.Store store) {
}
});
taskService.init();
}
@AfterClass
public static void shutdownService() {
var instance = RaftEngine.getInstance();
if (instance != null) {
instance.shutDown();
}
}
private static boolean deleteDir(File dir) {
if (dir.isDirectory()) {
for (File file : dir.listFiles()) {
deleteDir(file);
}
}
return dir.delete();
}
public static StoreNodeService getStoreNodeService() {
return storeNodeService;
}
public static PartitionService getPartitionService() {
return partitionService;
}
public static PDConfig getPdConfig() {
return pdConfig;
}
public static TaskScheduleService getTaskService() {
return taskService;
}
public static StoreMonitorDataService getStoreMonitorDataService() {
return storeMonitorDataService;
}
}

View File

@ -0,0 +1,156 @@
/*
* 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.hugegraph.pd.core;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.hugegraph.pd.PartitionService;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.grpc.MetaTask;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.pd.grpc.pulse.CleanPartition;
import org.apache.hugegraph.pd.grpc.pulse.CleanType;
import org.junit.Before;
import org.junit.Test;
public class PartitionServiceTest extends PDCoreTestBase {
private PartitionService service;
@Before
public void init() {
this.service = getPartitionService();
}
@Test
public void testCombinePartition() throws PDException {
buildEnv();
// 0, 1, 2 -> 0
// 3, 4, 5 -> 1
// 6, 7, 8 -> 2
// 9, 10, 11 -> 3
this.service.combinePartition(4);
var partition = this.service.getPartitionById("graph0", 0);
assertEquals(0, partition.getStartKey());
assertEquals(5462, partition.getEndKey());
var tasks = getStoreNodeService().getTaskInfoMeta().scanMoveTask("graph0");
assertEquals(11, tasks.size());
for (MetaTask.Task task : tasks) {
var newTask = task.toBuilder().setState(MetaTask.TaskState.Task_Success).build();
getTaskService().reportTask(newTask);
}
tasks = getStoreNodeService().getTaskInfoMeta().scanMoveTask("graph0");
assertEquals(0, tasks.size());
}
@Test
public void testCombinePartition2() throws PDException {
buildEnv();
// 0, 1, 2 -> 0
// 3, 4, 5 -> 1
// 6, 7, 8 -> 2
// 9, 10, 11 -> 3
this.service.combinePartition(4);
var partition = this.service.getPartitionById("graph0", 0);
assertEquals(0, partition.getStartKey());
assertEquals(5462, partition.getEndKey());
var tasks = getStoreNodeService().getTaskInfoMeta().scanMoveTask("graph0");
assertEquals(11, tasks.size());
for (MetaTask.Task task : tasks) {
var newTask = task.toBuilder().setState(MetaTask.TaskState.Task_Failure).build();
getTaskService().reportTask(newTask);
}
tasks = getStoreNodeService().getTaskInfoMeta().scanMoveTask("graph0");
assertEquals(0, tasks.size());
}
@Test
public void testHandleCleanTask() {
MetaTask.Task task = MetaTask.Task.newBuilder()
.setType(MetaTask.TaskType.Clean_Partition)
.setPartition(
Metapb.Partition.newBuilder().setGraphName("foo")
.setId(0).build())
.setCleanPartition(CleanPartition.newBuilder()
.setCleanType(
CleanType.CLEAN_TYPE_KEEP_RANGE)
.setDeletePartition(true)
.setKeyStart(0)
.setKeyEnd(10)
.build())
.build();
getTaskService().reportTask(task);
}
private void buildEnv() throws PDException {
var storeInfoMeta = getStoreNodeService().getStoreInfoMeta();
storeInfoMeta.updateStore(Metapb.Store.newBuilder()
.setId(99)
.setState(Metapb.StoreState.Up)
.build());
long lastId = 0;
for (int i = 0; i < 12; i++) {
Metapb.Shard shard = Metapb.Shard.newBuilder()
.setStoreId(99)
.setRole(Metapb.ShardRole.Leader)
.build();
Metapb.ShardGroup shardGroup = Metapb.ShardGroup.newBuilder()
.setId(i)
.setState(
Metapb.PartitionState.PState_Normal)
.addAllShards(List.of(shard))
.build();
storeInfoMeta.updateShardGroup(shardGroup);
var partitionShard = this.service.getPartitionByCode("graph0", lastId);
if (partitionShard != null) {
lastId = partitionShard.getPartition().getEndKey();
}
}
}
@Test
public void testPartitionHeartbeat() {
List<Metapb.Shard> shardList = new ArrayList<>();
shardList.add(Metapb.Shard.newBuilder().setStoreId(1).build());
shardList.add(Metapb.Shard.newBuilder().setStoreId(2).build());
shardList.add(Metapb.Shard.newBuilder().setStoreId(3).build());
shardList = new ArrayList<>(shardList);
Metapb.PartitionStats stats = Metapb.PartitionStats.newBuilder()
.addAllShard(shardList).build();
List<Metapb.Shard> shardList2 = new ArrayList<>(stats.getShardList());
Collections.shuffle(shardList2);
shardList2.forEach(shard -> System.out.println(shard.getStoreId()));
}
}

View File

@ -0,0 +1,79 @@
/*
* 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.hugegraph.pd.core;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.List;
import org.apache.hugegraph.pd.StoreMonitorDataService;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.junit.Before;
import org.junit.Test;
public class StoreMonitorDataServiceTest extends PDCoreTestBase {
private StoreMonitorDataService service;
@Before
public void init() {
this.service = getStoreMonitorDataService();
var store = getPdConfig().getStore();
store.setMonitorDataEnabled(true);
store.setMonitorDataInterval("1s");
getPdConfig().setStore(store);
}
@Test
public void test() throws InterruptedException, PDException {
long now = System.currentTimeMillis() / 1000;
for (int i = 0; i < 5; i++) {
this.service.saveMonitorData(genStats());
now = System.currentTimeMillis() / 1000;
Thread.sleep(1100);
}
assertTrue(this.service.getLatestStoreMonitorDataTimeStamp(1) == 0 ||
this.service.getLatestStoreMonitorDataTimeStamp(1) == now);
var data = this.service.getStoreMonitorData(1);
assertEquals(5, data.size());
assertNotNull(this.service.debugMonitorInfo(List.of(Metapb.RecordPair.newBuilder()
.setKey("key1")
.setValue(1)
.build())));
assertNotNull(this.service.getStoreMonitorDataText(1));
this.service.removeExpiredMonitorData(1, now + 1);
assertEquals(0, this.service.getStoreMonitorData(1).size());
}
private Metapb.StoreStats genStats() {
return Metapb.StoreStats.newBuilder()
.setStoreId(1)
.addSystemMetrics(
Metapb.RecordPair.newBuilder().setKey("key1").setValue(1)
.build())
.build();
}
}

View File

@ -0,0 +1,830 @@
/*
* 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.hugegraph.pd.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.apache.hugegraph.pd.PartitionService;
import org.apache.hugegraph.pd.StoreNodeService;
import org.apache.hugegraph.pd.StoreStatusListener;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.config.PDConfig;
import org.apache.hugegraph.pd.grpc.MetaTask;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
public class StoreServiceTest extends PDCoreTestBase {
private PDConfig config;
private StoreNodeService service;
@Before
public void setUp() {
this.config = getPdConfig();
this.service = new StoreNodeService(this.config);
}
@Test
public void testInit() {
// Setup
PDConfig pdConfig = getPdConfig();
final PartitionService partitionService = new PartitionService(pdConfig,
new StoreNodeService(
pdConfig));
// Run the test
this.service.init(partitionService);
// Verify the results
}
@Test
public void testIsOK() {
// Setup
// Run the test
final boolean result = this.service.isOK();
// Verify the results
assertThat(result).isTrue();
}
@Test
public void testRegister() throws Exception {
// Setup
final Metapb.Store store = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version").setState(
Metapb.StoreState.Unknown).setStartTimestamp(0L)
.setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build();
final Metapb.Store expectedResult = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress(
"raftAddress")
.addLabels(
Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version")
.setState(
Metapb.StoreState.Unknown)
.setStartTimestamp(0L)
.setDeployPath(
"deployPath")
.setLastHeartbeat(0L)
.setStats(
Metapb.StoreStats
.newBuilder()
.setStoreId(
0L)
.setPartitionCount(
0)
.addGraphStats(
Metapb.GraphStats
.newBuilder()
.setGraphName(
"value")
.setApproximateSize(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.build())
.setDataVersion(0)
.setCores(0)
.setDataPath("dataPath")
.build();
// Configure PDConfig.getInitialStoreMap(...).
final Map<String, String> stringStringMap = Map.ofEntries(
Map.entry("value", "value"));
// Run the test
final Metapb.Store result = this.service.register(store);
}
@Test
public void testGetStore() throws Exception {
// Setup
try {
Metapb.GraphStats stats = Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(
Metapb.ShardRole.None)
.build();
Metapb.StoreStats storeStats = Metapb.StoreStats.newBuilder()
.setStoreId(0L)
.setPartitionCount(
0)
.addGraphStats(
stats)
.build();
final Metapb.Store expectedResult = Metapb.Store.newBuilder()
.setId(0L)
.setAddress(
"address")
.setRaftAddress(
"raftAddress")
.addLabels(
Metapb.StoreLabel
.newBuilder()
.build())
.setVersion(
"version")
.setState(
Metapb.StoreState.Unknown)
.setStartTimestamp(
0L)
.setDeployPath(
"deployPath")
.setLastHeartbeat(
0L)
.setStats(
storeStats)
.setDataVersion(0)
.setCores(0)
.setDataPath(
"dataPath")
.build();
// Run the test
final Metapb.Store result = this.service.getStore(0L);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testUpdateStore() throws Exception {
// Setup
final Metapb.Store store = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version").setState(
Metapb.StoreState.Unknown).setStartTimestamp(0L)
.setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build();
final Metapb.Store expectedResult = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress(
"raftAddress")
.addLabels(
Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version")
.setState(
Metapb.StoreState.Unknown)
.setStartTimestamp(0L)
.setDeployPath(
"deployPath")
.setLastHeartbeat(0L)
.setStats(
Metapb.StoreStats
.newBuilder()
.setStoreId(
0L)
.setPartitionCount(
0)
.addGraphStats(
Metapb.GraphStats
.newBuilder()
.setGraphName(
"value")
.setApproximateSize(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.build())
.setDataVersion(0)
.setCores(0)
.setDataPath("dataPath")
.build();
// Configure PDConfig.getPartition(...).
final PDConfig.Partition partition = new PDConfig().new Partition();
partition.setTotalCount(0);
partition.setMaxShardsPerStore(0);
partition.setShardCount(0);
// Run the test
final Metapb.Store result = this.service.updateStore(store);
}
@Test
public void testStoreTurnoff() throws Exception {
// Setup
try {
final Metapb.Store store = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version").setState(
Metapb.StoreState.Unknown).setStartTimestamp(0L)
.setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build();
// Configure PDConfig.getPartition(...).
final PDConfig.Partition partition = new PDConfig().new Partition();
partition.setTotalCount(0);
partition.setMaxShardsPerStore(0);
partition.setShardCount(0);
// Run the test
this.service.storeTurnoff(store);
// Verify the results
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testGetStores1() throws Exception {
// Setup
final List<Metapb.Store> expectedResult = List.of(
Metapb.Store.newBuilder().setId(0L).setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel.newBuilder().build())
.setVersion("version")
.setState(Metapb.StoreState.Unknown)
.setStartTimestamp(0L).setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build());
// Run the test
final List<Metapb.Store> result = this.service.getStores();
}
@Test
public void testGetStores2() throws Exception {
// Setup
final List<Metapb.Store> expectedResult = List.of(
Metapb.Store.newBuilder().setId(0L).setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel.newBuilder().build())
.setVersion("version")
.setState(Metapb.StoreState.Unknown)
.setStartTimestamp(0L).setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build());
// Run the test
final List<Metapb.Store> result = this.service.getStores("graphName");
}
@Test
public void testGetStoreStatus() throws Exception {
// Setup
final List<Metapb.Store> expectedResult = List.of(
Metapb.Store.newBuilder().setId(0L).setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel.newBuilder().build())
.setVersion("version")
.setState(Metapb.StoreState.Unknown)
.setStartTimestamp(0L).setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build());
// Run the test
final List<Metapb.Store> result = this.service.getStoreStatus(false);
}
@Test
public void testGetShardGroups() throws Exception {
// Setup
final List<Metapb.ShardGroup> expectedResult = List.of(
Metapb.ShardGroup.newBuilder().setId(0).addShards(
Metapb.Shard.newBuilder().setStoreId(0L)
.setRole(Metapb.ShardRole.None).build())
.setState(Metapb.PartitionState.PState_None)
.build());
// Run the test
final List<Metapb.ShardGroup> result = this.service.getShardGroups();
}
@Test
public void testGetShardGroup() throws Exception {
// Setup
final Metapb.ShardGroup expectedResult = Metapb.ShardGroup.newBuilder()
.setId(0)
.addShards(
Metapb.Shard
.newBuilder()
.setStoreId(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.setState(
Metapb.PartitionState.PState_None)
.build();
// Run the test
final Metapb.ShardGroup result = this.service.getShardGroup(0);
// Verify the results
}
@Test
public void testGetShardGroupsByStore() throws Exception {
// Setup
final List<Metapb.ShardGroup> expectedResult = List.of(
Metapb.ShardGroup.newBuilder().setId(0).addShards(
Metapb.Shard.newBuilder().setStoreId(0L)
.setRole(Metapb.ShardRole.None).build())
.setState(Metapb.PartitionState.PState_None)
.build());
// Run the test
final List<Metapb.ShardGroup> result = this.service.getShardGroupsByStore(
0L);
}
@Test
public void testGetActiveStores1() throws Exception {
// Setup
final List<Metapb.Store> expectedResult = List.of(
Metapb.Store.newBuilder().setId(0L).setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel.newBuilder().build())
.setVersion("version")
.setState(Metapb.StoreState.Unknown)
.setStartTimestamp(0L).setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build());
// Run the test
final List<Metapb.Store> result = this.service.getActiveStores("graphName");
// Verify the results
}
@Test
public void testGetActiveStores1ThrowsPDException() {
try {
List<Metapb.Store> stores = this.service.getActiveStores();
assertThat(stores.size()).isEqualTo(0);
} catch (Exception e) {
e.printStackTrace();
}
}
@Ignore // state is Pending instead of Tombstone
@Test
public void testGetTombStores() throws Exception {
// Setup
final List<Metapb.Store> storeList = List.of(
Metapb.Store.newBuilder().setId(0L).setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel.newBuilder().build())
.setVersion("version")
.setState(Metapb.StoreState.Tombstone)
.setStartTimestamp(0L).setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build());
this.service.register(storeList.get(0));
// Run the test
final List<Metapb.Store> result = this.service.getTombStores();
// Verify the results
assertThat(result.size()).isEqualTo(1);
this.service.removeStore(result.get(0).getId());
List<Metapb.Store> stores = this.service.getStores();
assertThat(stores.size()).isEqualTo(0);
}
@Test
public void testAllocShards() throws Exception {
// Setup
try {
final Metapb.Graph graph = Metapb.Graph.newBuilder()
.setGraphName("graphName")
.setGraphState(
Metapb.GraphState
.newBuilder()
.setMode(
Metapb.GraphMode.ReadWrite)
.setReason(
Metapb.GraphModeReason.Quota)
.build())
.build();
final List<Metapb.Shard> expectedResult = List.of(
Metapb.Shard.newBuilder().setStoreId(0L)
.setRole(Metapb.ShardRole.None).build());
// Configure PDConfig.getPartition(...).
final PDConfig.Partition partition = new PDConfig().new Partition();
partition.setTotalCount(0);
partition.setMaxShardsPerStore(0);
partition.setShardCount(0);
// Run the test
final List<Metapb.Shard> result = this.service.allocShards(graph, 0);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testReallocShards() throws Exception {
// Setup
try {
final Metapb.ShardGroup shardGroup = Metapb.ShardGroup.newBuilder()
.setId(0)
.addShards(
Metapb.Shard
.newBuilder()
.setStoreId(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.setState(
Metapb.PartitionState.PState_None)
.build();
final List<Metapb.Shard> expectedResult = List.of(
Metapb.Shard.newBuilder().setStoreId(0L)
.setRole(Metapb.ShardRole.None).build());
// Configure PDConfig.getPartition(...).
final PDConfig.Partition partition = new PDConfig().new Partition();
partition.setTotalCount(0);
partition.setMaxShardsPerStore(0);
partition.setShardCount(0);
when(this.config.getPartition()).thenReturn(partition);
// Run the test
final List<Metapb.Shard> result = this.service.reallocShards(shardGroup);
// Verify the results
assertThat(result).isEqualTo(expectedResult);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testUpdateShardGroup() {
try {
final List<Metapb.Shard> shards = List.of(
Metapb.Shard.newBuilder().setStoreId(0L)
.setRole(Metapb.ShardRole.None).build());
// Run the test
this.service.updateShardGroup(0, shards, 0, 0);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testUpdateShardGroupState() throws Exception {
try {
this.service.updateShardGroupState(0, Metapb.PartitionState.PState_None);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testHeartBeat() throws Exception {
// Setup
try {
final Metapb.StoreStats storeStats = Metapb.StoreStats.newBuilder()
.setStoreId(
0L)
.setPartitionCount(
0)
.addGraphStats(
Metapb.GraphStats
.newBuilder()
.setGraphName(
"value")
.setApproximateSize(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.build();
final Metapb.ClusterStats expectedResult = Metapb.ClusterStats
.newBuilder().setState(Metapb.ClusterState.Cluster_OK)
.setMessage("message").setTimestamp(0L).build();
when(this.config.getMinStoreCount()).thenReturn(0);
// Configure PDConfig.getPartition(...).
final PDConfig.Partition partition = new PDConfig().new Partition();
partition.setTotalCount(0);
partition.setMaxShardsPerStore(0);
partition.setShardCount(0);
when(this.config.getPartition()).thenReturn(partition);
// Run the test
final Metapb.ClusterStats result = this.service.heartBeat(storeStats);
// Verify the results
assertThat(result).isEqualTo(expectedResult);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testUpdateClusterStatus1() {
// Setup
final Metapb.ClusterStats expectedResult = Metapb.ClusterStats
.newBuilder().setState(Metapb.ClusterState.Cluster_OK)
.setMessage("message").setTimestamp(0L).build();
// Run the test
final Metapb.ClusterStats result = this.service.updateClusterStatus(
Metapb.ClusterState.Cluster_OK);
}
@Test
public void testUpdateClusterStatus2() {
// Setup
final Metapb.ClusterStats expectedResult = Metapb.ClusterStats
.newBuilder().setState(Metapb.ClusterState.Cluster_OK)
.setMessage("message").setTimestamp(0L).build();
// Run the test
final Metapb.ClusterStats result = this.service.updateClusterStatus(
Metapb.PartitionState.PState_None);
}
@Test
public void testCheckStoreStatus() {
// Setup
// Run the test
this.service.checkStoreStatus();
// Verify the results
}
@Test
public void testAddStatusListener() {
// Setup
final StoreStatusListener mockListener = mock(
StoreStatusListener.class);
// Run the test
this.service.addStatusListener(mockListener);
// Verify the results
}
@Test
public void testOnStoreStatusChanged() {
// Setup
final Metapb.Store store = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress("raftAddress")
.addLabels(Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version").setState(
Metapb.StoreState.Unknown).setStartTimestamp(0L)
.setDeployPath("deployPath")
.setLastHeartbeat(0L).setStats(
Metapb.StoreStats.newBuilder().setStoreId(0L)
.setPartitionCount(0).addGraphStats(
Metapb.GraphStats.newBuilder()
.setGraphName("value")
.setApproximateSize(0L)
.setRole(Metapb.ShardRole.None)
.build()).build())
.setDataVersion(0).setCores(0)
.setDataPath("dataPath").build();
// Verify the results
}
@Test
public void testOnShardGroupSplit() {
// Setup
final Metapb.ShardGroup shardGroup = Metapb.ShardGroup.newBuilder()
.setId(0)
.addShards(
Metapb.Shard
.newBuilder()
.setStoreId(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.setState(
Metapb.PartitionState.PState_None)
.build();
final List<Metapb.ShardGroup> newShardGroups = List.of(
Metapb.ShardGroup.newBuilder().setId(0).addShards(
Metapb.Shard.newBuilder().setStoreId(0L)
.setRole(Metapb.ShardRole.None).build())
.setState(Metapb.PartitionState.PState_None)
.build());
final Consumer<MetaTask.Task> mockTask = mock(Consumer.class);
// Verify the results
}
@Ignore // active stores are fewer than min store count in pd config
@Test
public void testCheckStoreCanOffline() {
// Setup
final Metapb.Store currentStore = Metapb.Store.newBuilder().setId(0L)
.setAddress("address")
.setRaftAddress(
"raftAddress")
.addLabels(
Metapb.StoreLabel
.newBuilder()
.build())
.setVersion("version")
.setState(
Metapb.StoreState.Unknown)
.setStartTimestamp(0L)
.setDeployPath(
"deployPath")
.setLastHeartbeat(0L)
.setStats(
Metapb.StoreStats
.newBuilder()
.setStoreId(
0L)
.setPartitionCount(
0)
.addGraphStats(
Metapb.GraphStats
.newBuilder()
.setGraphName(
"value")
.setApproximateSize(
0L)
.setRole(
Metapb.ShardRole.None)
.build())
.build())
.setDataVersion(0)
.setCores(0)
.setDataPath("dataPath")
.build();
// Run the test
final boolean result = this.service.checkStoreCanOffline(currentStore);
// Verify the results
assertThat(result).isTrue();
}
@Test
public void testShardGroupsDbCompaction() throws Exception {
// Setup
// Run the test
try {
this.service.shardGroupsDbCompaction(0, "tableName");
} catch (Exception e) {
e.printStackTrace();
}
// Verify the results
}
@Test
public void testGetQuota() throws Exception {
// Setup
// Run the test
try {
this.service.getQuota();
} catch (Exception e) {
e.printStackTrace();
}
}
// migrated from StoreNodeServiceNewTest
@Test
public void testRemoveShardGroup() throws PDException {
for (int i = 0; i < 12; i++) {
Metapb.ShardGroup group = Metapb.ShardGroup.newBuilder()
.setId(i)
.setState(
Metapb.PartitionState.PState_Offline)
.build();
this.service.getStoreInfoMeta().updateShardGroup(group);
}
this.service.deleteShardGroup(11);
this.service.deleteShardGroup(10);
assertEquals(10, getPdConfig().getConfigService().getPDConfig().getPartitionCount());
// restore
getPdConfig().getConfigService().setPartitionCount(12);
}
}

View File

@ -0,0 +1,117 @@
/*
* 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.hugegraph.pd.core;
import static org.junit.Assert.assertFalse;
import java.util.ArrayList;
import java.util.List;
import org.apache.hugegraph.pd.TaskScheduleService;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.junit.Before;
import org.junit.Test;
public class TaskScheduleServiceTest extends PDCoreTestBase {
TaskScheduleService service;
@Before
public void init() {
this.service = getTaskService();
}
// TODO
public void testStoreOffline() {
}
// TODO
public void testPatrolStores() {
}
// TODO
public void testPatrolPartitions() {
}
// TODO
public void testBalancePartitionShard() {
}
@Test
public void testBalancePartitionLeader() throws PDException {
var list = new ArrayList<Metapb.Partition>();
for (int i = 0; i < 6; i++) {
getStoreNodeService().getStoreInfoMeta().updateShardGroup(genShardGroup(i));
list.add(genPartition(i));
}
getPdConfig().getPartition().setShardCount(3);
getPartitionService().updatePartition(list);
var rst = this.service.balancePartitionLeader(true);
assertFalse(rst.isEmpty());
// recover
getPdConfig().getPartition().setShardCount(1);
getStoreNodeService().getStoreInfoMeta().removeAll();
}
// TODO
public void testSplitPartition() {
}
// TODO
public void testSplitPartition2() {
}
// TODO
public void testCanAllPartitionsMovedOut() {
}
private Metapb.ShardGroup genShardGroup(int groupId) {
return Metapb.ShardGroup.newBuilder()
.setId(groupId)
.addAllShards(genShards())
.build();
}
private Metapb.Partition genPartition(int groupId) {
return Metapb.Partition.newBuilder()
.setId(groupId)
.setState(Metapb.PartitionState.PState_Normal)
.setGraphName("graph1")
.setStartKey(groupId * 10L)
.setEndKey(groupId * 10L + 10)
.build();
}
private List<Metapb.Shard> genShards() {
return List.of(
Metapb.Shard.newBuilder().setStoreId(1).setRole(Metapb.ShardRole.Leader).build(),
Metapb.Shard.newBuilder().setStoreId(2).setRole(Metapb.ShardRole.Follower).build(),
Metapb.Shard.newBuilder().setStoreId(3).setRole(Metapb.ShardRole.Follower).build());
}
}

View File

@ -0,0 +1,226 @@
/*
* 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.hugegraph.pd.core.meta;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertArrayEquals;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.pd.meta.MetadataKeyHelper;
import org.junit.Test;
public class MetadataKeyHelperTest {
@Test
public void testMoveTaskKey() {
var key = MetadataKeyHelper.getMoveTaskKey("foo", 0, 1);
assertArrayEquals(key, "TASK_MOVE/foo/0/1".getBytes());
var key2 = MetadataKeyHelper.getMoveTaskPrefix("foo");
assertArrayEquals(key2, "TASK_MOVE/foo".getBytes());
}
@Test
public void testGetStoreInfoKey() {
assertThat(MetadataKeyHelper.getStoreInfoKey(0L)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetActiveStoreKey() {
assertThat(MetadataKeyHelper.getActiveStoreKey(0L)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetActiveStorePrefix() {
assertThat(MetadataKeyHelper.getActiveStorePrefix()).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetStorePrefix() {
assertThat(MetadataKeyHelper.getStorePrefix()).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetStoreStatusKey() {
assertThat(MetadataKeyHelper.getStoreStatusKey(0L)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetShardGroupKey() {
assertThat(MetadataKeyHelper.getShardGroupKey(0L)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetShardGroupPrefix() {
assertThat(MetadataKeyHelper.getShardGroupPrefix()).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetPartitionKey() {
assertThat(MetadataKeyHelper.getPartitionKey("graphName", 0)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetPartitionPrefix() {
assertThat(MetadataKeyHelper.getPartitionPrefix("graphName")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetShardKey() {
assertThat(MetadataKeyHelper.getShardKey(0L, 0)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetShardPrefix() {
assertThat(MetadataKeyHelper.getShardPrefix(0L)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetGraphKey() {
assertThat(MetadataKeyHelper.getGraphKey("graphName")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetGraphPrefix() {
assertThat(MetadataKeyHelper.getGraphPrefix()).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetPartitionStatusKey() {
assertThat(MetadataKeyHelper.getPartitionStatusKey("graphName",
0)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetPartitionStatusPrefixKey() {
assertThat(MetadataKeyHelper.getPartitionStatusPrefixKey(
"graphName")).contains(MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetGraphSpaceKey() {
assertThat(MetadataKeyHelper.getGraphSpaceKey("graphSpace")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetPdConfigKey() {
assertThat(MetadataKeyHelper.getPdConfigKey("configKey")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetQueueItemPrefix() {
assertThat(MetadataKeyHelper.getQueueItemPrefix()).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetQueueItemKey() {
assertThat(MetadataKeyHelper.getQueueItemKey("itemId")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetSpitTaskKey() {
assertThat(MetadataKeyHelper.getSplitTaskKey("graphName", 0)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetSpitTaskPrefix() {
assertThat(MetadataKeyHelper.getSplitTaskPrefix("graph0")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetLogKey() {
// Setup
final Metapb.LogRecord record = Metapb.LogRecord.newBuilder()
.setAction("value")
.setTimestamp(0L)
.build();
// Run the test
final byte[] result = MetadataKeyHelper.getLogKey(record);
// Verify the results
assertThat(result).contains(MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetLogKeyPrefix() {
assertThat(MetadataKeyHelper.getLogKeyPrefix("action", 0L)).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetKVPrefix() {
assertThat(MetadataKeyHelper.getKVPrefix("prefix", "key")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetKVTTLPrefix() {
assertThat(MetadataKeyHelper.getKVTTLPrefix("ttlPrefix", "prefix",
"key")).contains(
MetadataKeyHelper.getDelimiter());
}
@Test
public void testGetKVWatchKeyPrefix1() {
assertThat(
MetadataKeyHelper.getKVWatchKeyPrefix("key", "watchDelimiter",
0L)).contains(
String.valueOf(MetadataKeyHelper.getDelimiter()));
}
@Test
public void testGetKVWatchKeyPrefix2() {
assertThat(MetadataKeyHelper.getKVWatchKeyPrefix("key",
"watchDelimiter")).contains(
String.valueOf(MetadataKeyHelper.getDelimiter()));
}
@Test
public void testGetDelimiter() {
assertThat(MetadataKeyHelper.getDelimiter()).isEqualTo('/');
}
@Test
public void testGetStringBuilderHelper() {
try {
MetadataKeyHelper.getStringBuilderHelper();
} catch (Exception e) {
}
}
}

View File

@ -0,0 +1,113 @@
/*
* 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.hugegraph.pd.core.store;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
import org.apache.commons.io.FileUtils;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.config.PDConfig;
import org.apache.hugegraph.pd.store.HgKVStore;
import org.apache.hugegraph.pd.store.HgKVStoreImpl;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
public class HgKVStoreImplTest {
private static final String testPath = "tmp/test";
private static PDConfig pdConfig;
@BeforeClass
public static void init() throws IOException {
File testFile = new File(testPath);
if (testFile.exists()) {
FileUtils.deleteDirectory(testFile);
}
FileUtils.forceMkdir(testFile);
pdConfig = new PDConfig() {{
setDataPath(testPath);
}};
}
@Test
public void Test() throws PDException {
HgKVStore kvStore = new HgKVStoreImpl();
kvStore.init(pdConfig);
{
byte[] key = "hello".getBytes();
byte[] value = "pd".getBytes();
kvStore.put(key, value);
}
for (int i = 0; i < 100; i++) {
byte[] key = String.format("k%03d", i).getBytes();
byte[] value = ("value" + i).getBytes();
kvStore.put(key, value);
}
Assert.assertEquals(100, kvStore.scanPrefix("k".getBytes()).size());
kvStore.removeByPrefix("k".getBytes());
Assert.assertEquals(0, kvStore.scanPrefix("k".getBytes()).size());
kvStore.close();
}
@Test
public void TestSnapshot() throws PDException {
HgKVStore kvStore = new HgKVStoreImpl();
kvStore.init(pdConfig);
// put 100 data
for (int i = 0; i < 100; i++) {
byte[] key = String.format("k%03d", i).getBytes();
byte[] value = ("value" + i).getBytes();
kvStore.put(key, value);
}
Assert.assertEquals(100, kvStore.scanPrefix("k".getBytes()).size());
// save snapshot
String snapshotPath = Paths.get(testPath, "snapshot").toString();
kvStore.saveSnapshot(snapshotPath);
// put another 100 data
for (int i = 100; i < 200; i++) {
byte[] key = String.format("k%03d", i).getBytes();
byte[] value = ("value" + i).getBytes();
kvStore.put(key, value);
}
Assert.assertEquals(200, kvStore.scanPrefix("k".getBytes()).size());
// load snapshot
kvStore.loadSnapshot(snapshotPath);
Assert.assertEquals(100, kvStore.scanPrefix("k".getBytes()).size());
// put another 100 data
for (int i = 100; i < 200; i++) {
byte[] key = String.format("k%03d", i).getBytes();
byte[] value = ("value" + i).getBytes();
kvStore.put(key, value);
}
Assert.assertEquals(200, kvStore.scanPrefix("k".getBytes()).size());
kvStore.close();
}
}

View File

@ -15,12 +15,14 @@
* limitations under the License.
*/
package org.apache.hugegraph.pd.common;
package org.apache.hugegraph.pd.grpc;
import org.apache.hugegraph.pd.common.Useless;
import org.junit.After;
import org.junit.BeforeClass;
public class BaseCommonTest {
@Useless("empty now")
public class BaseGrpcTest {
@BeforeClass
public static void init() {
@ -31,4 +33,5 @@ public class BaseCommonTest {
public void teardown() {
// pass
}
}

View File

@ -0,0 +1,33 @@
/*
* 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.hugegraph.pd.grpc;
import org.apache.hugegraph.pd.common.Useless;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
import lombok.extern.slf4j.Slf4j;
@Useless("empty now")
@RunWith(Suite.class)
@Suite.SuiteClasses({
})
@Slf4j
public class GrpcSuiteTest {
}

View File

@ -0,0 +1,41 @@
/*
* 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.hugegraph.pd.rest;
import java.net.http.HttpClient;
import org.junit.After;
import org.junit.BeforeClass;
public class BaseServerTest {
protected static HttpClient client;
protected static String pdRestAddr;
@BeforeClass
public static void init() {
client = HttpClient.newHttpClient();
pdRestAddr = "http://127.0.0.1:8620";
}
@After
public void teardown() {
// pass
}
}

View File

@ -0,0 +1,32 @@
/*
* 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.hugegraph.pd.rest;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
import lombok.extern.slf4j.Slf4j;
@RunWith(Suite.class)
@Suite.SuiteClasses({
RestApiTest.class,
})
@Slf4j
public class PDRestSuiteTest {
}

View File

@ -0,0 +1,121 @@
/*
* 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.hugegraph.pd.rest;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import org.json.JSONException;
import org.json.JSONObject;
import org.junit.Test;
public class RestApiTest extends BaseServerTest {
@Test
public void testQueryClusterInfo() throws URISyntaxException, IOException, InterruptedException,
JSONException {
String url = pdRestAddr + "/v1/cluster";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
JSONObject obj = new JSONObject(response.body());
assert obj.getInt("status") == 0;
}
@Test
public void testQueryClusterMembers() throws URISyntaxException, IOException,
InterruptedException, JSONException {
String url = pdRestAddr + "/v1/members";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
JSONObject obj = new JSONObject(response.body());
assert obj.getInt("status") == 0;
}
@Test
public void testQueryStoresInfo() throws URISyntaxException, IOException, InterruptedException,
JSONException {
String url = pdRestAddr + "/v1/stores";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
JSONObject obj = new JSONObject(response.body());
assert obj.getInt("status") == 0;
}
@Test
public void testQueryGraphsInfo() throws IOException, InterruptedException, JSONException,
URISyntaxException {
String url = pdRestAddr + "/v1/graphs";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
JSONObject obj = new JSONObject(response.body());
assert obj.getInt("status") == 0;
}
@Test
public void testQueryPartitionsInfo() throws IOException, InterruptedException, JSONException,
URISyntaxException {
String url = pdRestAddr + "/v1/highLevelPartitions";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
JSONObject obj = new JSONObject(response.body());
assert obj.getInt("status") == 0;
}
@Test
public void testQueryDebugPartitionsInfo() throws URISyntaxException, IOException,
InterruptedException {
String url = pdRestAddr + "/v1/partitions";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
assert response.statusCode() == 200;
}
@Test
public void testQueryShards() throws URISyntaxException, IOException, InterruptedException,
JSONException {
String url = pdRestAddr + "/v1/shards";
HttpRequest request = HttpRequest.newBuilder()
.uri(new URI(url))
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
JSONObject obj = new JSONObject(response.body());
assert obj.getInt("status") == 0;
}
}

View File

@ -38,8 +38,7 @@
<module>hg-pd-test</module>
<module>hg-pd-core</module>
<module>hg-pd-service</module>
<!-- TODO: uncomment later -->
<!-- <module>hg-pd-dist</module> -->
<module>hg-pd-dist</module>
</modules>
<properties>
@ -179,5 +178,77 @@
</plugins>
</build>
</profile>
<profile>
<id>pd-core-test</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.20</version>
<executions>
<execution>
<id>pd-core-test</id>
<goals>
<goal>test</goal>
</goals>
<phase>test</phase>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>pd-client-test</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.20</version>
<executions>
<execution>
<id>pd-client-test</id>
<goals>
<goal>test</goal>
</goals>
<phase>test</phase>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>pd-rest-test</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.20</version>
<executions>
<execution>
<id>pd-rest-test</id>
<goals>
<goal>test</goal>
</goals>
<phase>test</phase>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

View File

@ -0,0 +1,26 @@
#!/bin/bash
#
# 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.
#
set -ev
HOME_DIR=$(pwd)
PD_DIR=$HOME_DIR/hugegraph-pd/dist/hugegraph-pd-1.3.0
pushd $PD_DIR
. bin/start-hugegraph-pd.sh
sleep 10
popd