Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
Shardingsphere
提交
689dce33
Shardingsphere
项目概览
apache
/
Shardingsphere
通知
56
Star
3
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
Shardingsphere
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
689dce33
编写于
3月 22, 2019
作者:
T
terrymanu
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
remove sharding-orchestration-reg-etcd for grpc license reason
上级
da082d2a
变更
14
展开全部
隐藏空白更改
内联
并排
Showing
14 changed file
with
0 addition
and
1744 deletion
+0
-1744
pom.xml
pom.xml
+0
-47
sharding-orchestration/sharding-orchestration-reg/pom.xml
sharding-orchestration/sharding-orchestration-reg/pom.xml
+0
-1
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/pom.xml
...orchestration-reg/sharding-orchestration-reg-etcd/pom.xml
+0
-53
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/EtcdRegistryCenter.java
...dingsphere/orchestration/reg/etcd/EtcdRegistryCenter.java
+0
-206
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/channel/EtcdChannelFactory.java
...tration/reg/etcd/internal/channel/EtcdChannelFactory.java
+0
-59
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/channel/EtcdNameSolverFactory.java
...tion/reg/etcd/internal/channel/EtcdNameSolverFactory.java
+0
-106
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/keepalive/KeepAlive.java
.../orchestration/reg/etcd/internal/keepalive/KeepAlive.java
+0
-145
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/retry/EtcdRetryEngine.java
...rchestration/reg/etcd/internal/retry/EtcdRetryEngine.java
+0
-66
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/watcher/EtcdWatchStreamObserver.java
...on/reg/etcd/internal/watcher/EtcdWatchStreamObserver.java
+0
-73
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/proto/etcd/auth/authpb/auth.proto
...ation-reg-etcd/src/main/proto/etcd/auth/authpb/auth.proto
+0
-45
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/proto/etcd/etcdserver/etcdserverpb/rpc.proto
...tcd/src/main/proto/etcd/etcdserver/etcdserverpb/rpc.proto
+0
-861
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/proto/etcd/mvcc/mvccpb/kv.proto
...tration-reg-etcd/src/main/proto/etcd/mvcc/mvccpb/kv.proto
+0
-59
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.orchestration.reg.api.RegistryCenter
...pache.shardingsphere.orchestration.reg.api.RegistryCenter
+0
-18
shardingsphere-distribution/shardingsphere-basic-distribution/pom.xml
...re-distribution/shardingsphere-basic-distribution/pom.xml
+0
-5
未找到文件。
pom.xml
浏览文件 @
689dce33
...
...
@@ -67,10 +67,6 @@
<atomikos.version>
4.0.6
</atomikos.version>
<curator.version>
2.10.0
</curator.version>
<grpc.version>
1.7.0
</grpc.version>
<protobuf.version>
3.4.0
</protobuf.version>
<guava-retrying.version>
2.0.0
</guava-retrying.version>
<opentracing.version>
0.30.0
</opentracing.version>
<lombok.version>
1.16.4
</lombok.version>
...
...
@@ -119,7 +115,6 @@
<jdepend-maven-plugin.version>
2.0
</jdepend-maven-plugin.version>
<taglist-maven-plugin.version>
2.4
</taglist-maven-plugin.version>
<maven-gpg-plugin.version>
1.6
</maven-gpg-plugin.version>
<protobuf-maven-plugin.version>
0.5.0
</protobuf-maven-plugin.version>
<os-maven-plugin.version>
1.5.0.Final
</os-maven-plugin.version>
<docker-maven-plugin.version>
0.4.14
</docker-maven-plugin.version>
<apache-rat-plugin.version>
0.12
</apache-rat-plugin.version>
...
...
@@ -222,26 +217,6 @@
<artifactId>
curator-recipes
</artifactId>
<version>
${curator.version}
</version>
</dependency>
<dependency>
<groupId>
io.grpc
</groupId>
<artifactId>
grpc-netty
</artifactId>
<version>
${grpc.version}
</version>
</dependency>
<dependency>
<groupId>
io.grpc
</groupId>
<artifactId>
grpc-protobuf
</artifactId>
<version>
${grpc.version}
</version>
</dependency>
<dependency>
<groupId>
io.grpc
</groupId>
<artifactId>
grpc-stub
</artifactId>
<version>
${grpc.version}
</version>
</dependency>
<dependency>
<groupId>
com.github.rholder
</groupId>
<artifactId>
guava-retrying
</artifactId>
<version>
${guava-retrying.version}
</version>
</dependency>
<dependency>
<groupId>
io.opentracing
</groupId>
...
...
@@ -685,24 +660,6 @@
</execution>
</executions>
</plugin>
<plugin>
<groupId>
org.xolstice.maven.plugins
</groupId>
<artifactId>
protobuf-maven-plugin
</artifactId>
<version>
${protobuf-maven-plugin.version}
</version>
<configuration>
<protocArtifact>
com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}
</protocArtifact>
<pluginId>
grpc-java
</pluginId>
<pluginArtifact>
io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}
</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>
compile
</goal>
<goal>
compile-custom
</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>
org.eluder.coveralls
</groupId>
<artifactId>
coveralls-maven-plugin
</artifactId>
...
...
@@ -733,11 +690,7 @@
</ignoreMethodAnnotations>
<excludes>
<exclude>
org/apache/shardingsphere/core/parse/antlr/autogen/*.class
</exclude>
<exclude>
authpb/*.class
</exclude>
<exclude>
mvccpb/*.class
</exclude>
<exclude>
etcdserverpb/*.class
</exclude>
<exclude>
org/apache/shardingsphere/orchestration/reg/zookeeper/*.class
</exclude>
<exclude>
org/apache/shardingsphere/orchestration/reg/etcd/*.class
</exclude>
<exclude>
org/apache/shardingsphere/**/*Test.class
</exclude>
<exclude>
org/apache/shardingsphere/**/Test*.class
</exclude>
</excludes>
...
...
sharding-orchestration/sharding-orchestration-reg/pom.xml
浏览文件 @
689dce33
...
...
@@ -30,6 +30,5 @@
<modules>
<module>
sharding-orchestration-reg-api
</module>
<module>
sharding-orchestration-reg-zookeeper-curator
</module>
<module>
sharding-orchestration-reg-etcd
</module>
</modules>
</project>
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/pom.xml
已删除
100644 → 0
浏览文件 @
da082d2a
<?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=
"http://maven.apache.org/POM/4.0.0"
xmlns:xsi=
"http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation=
"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
>
<modelVersion>
4.0.0
</modelVersion>
<parent>
<groupId>
org.apache.shardingsphere
</groupId>
<artifactId>
sharding-orchestration-reg
</artifactId>
<version>
4.0.0-RC1-SNAPSHOT
</version>
</parent>
<artifactId>
sharding-orchestration-reg-etcd
</artifactId>
<name>
${project.artifactId}
</name>
<dependencies>
<dependency>
<groupId>
org.apache.shardingsphere
</groupId>
<artifactId>
sharding-orchestration-reg-api
</artifactId>
<version>
${project.version}
</version>
</dependency>
<dependency>
<groupId>
io.grpc
</groupId>
<artifactId>
grpc-netty
</artifactId>
</dependency>
<dependency>
<groupId>
io.grpc
</groupId>
<artifactId>
grpc-protobuf
</artifactId>
</dependency>
<dependency>
<groupId>
io.grpc
</groupId>
<artifactId>
grpc-stub
</artifactId>
</dependency>
<dependency>
<groupId>
com.github.rholder
</groupId>
<artifactId>
guava-retrying
</artifactId>
</dependency>
</dependencies>
</project>
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/EtcdRegistryCenter.java
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.shardingsphere.orchestration.reg.etcd
;
import
com.google.common.base.Optional
;
import
com.google.common.base.Splitter
;
import
com.google.protobuf.ByteString
;
import
etcdserverpb.KVGrpc
;
import
etcdserverpb.KVGrpc.KVFutureStub
;
import
etcdserverpb.LeaseGrpc
;
import
etcdserverpb.LeaseGrpc.LeaseFutureStub
;
import
etcdserverpb.Rpc.LeaseGrantRequest
;
import
etcdserverpb.Rpc.PutRequest
;
import
etcdserverpb.Rpc.RangeRequest
;
import
etcdserverpb.Rpc.RangeResponse
;
import
etcdserverpb.Rpc.WatchCreateRequest
;
import
etcdserverpb.Rpc.WatchRequest
;
import
etcdserverpb.WatchGrpc
;
import
etcdserverpb.WatchGrpc.WatchStub
;
import
io.grpc.Channel
;
import
mvccpb.Kv.KeyValue
;
import
org.apache.shardingsphere.orchestration.reg.api.RegistryCenter
;
import
org.apache.shardingsphere.orchestration.reg.api.RegistryCenterConfiguration
;
import
org.apache.shardingsphere.orchestration.reg.etcd.internal.channel.EtcdChannelFactory
;
import
org.apache.shardingsphere.orchestration.reg.etcd.internal.keepalive.KeepAlive
;
import
org.apache.shardingsphere.orchestration.reg.etcd.internal.retry.EtcdRetryEngine
;
import
org.apache.shardingsphere.orchestration.reg.etcd.internal.watcher.EtcdWatchStreamObserver
;
import
org.apache.shardingsphere.orchestration.reg.exception.RegistryCenterException
;
import
org.apache.shardingsphere.orchestration.reg.listener.DataChangedEventListener
;
import
java.util.ArrayList
;
import
java.util.Arrays
;
import
java.util.Collections
;
import
java.util.List
;
import
java.util.concurrent.Callable
;
import
java.util.concurrent.ExecutionException
;
import
java.util.concurrent.TimeUnit
;
import
java.util.concurrent.TimeoutException
;
/**
* Etcd based registry center.
*
* @author junxiong
*/
public
final
class
EtcdRegistryCenter
implements
RegistryCenter
{
private
RegistryCenterConfiguration
config
;
private
EtcdRetryEngine
etcdRetryEngine
;
private
KVFutureStub
kvStub
;
private
LeaseFutureStub
leaseStub
;
private
WatchStub
watchStub
;
private
KeepAlive
keepAlive
;
@Override
public
void
init
(
final
RegistryCenterConfiguration
config
)
{
this
.
config
=
config
;
etcdRetryEngine
=
new
EtcdRetryEngine
(
config
);
Channel
channel
=
EtcdChannelFactory
.
getInstance
(
Splitter
.
on
(
','
).
trimResults
().
splitToList
(
config
.
getServerLists
()));
kvStub
=
KVGrpc
.
newFutureStub
(
channel
);
leaseStub
=
LeaseGrpc
.
newFutureStub
(
channel
);
watchStub
=
WatchGrpc
.
newStub
(
channel
);
keepAlive
=
new
KeepAlive
(
channel
,
config
.
getTimeToLiveSeconds
());
}
@Override
public
String
get
(
final
String
key
)
{
final
RangeRequest
request
=
RangeRequest
.
newBuilder
().
setKey
(
ByteString
.
copyFromUtf8
(
key
)).
build
();
return
etcdRetryEngine
.
execute
(
new
Callable
<
String
>()
{
@Override
public
String
call
()
throws
InterruptedException
,
ExecutionException
,
TimeoutException
{
RangeResponse
response
=
kvStub
.
range
(
request
).
get
(
config
.
getOperationTimeoutMilliseconds
(),
TimeUnit
.
MILLISECONDS
);
return
response
.
getKvsCount
()
>
0
?
response
.
getKvs
(
0
).
getValue
().
toStringUtf8
()
:
null
;
}
}).
orNull
();
}
@Override
public
String
getDirectly
(
final
String
key
)
{
return
get
(
key
);
}
@Override
public
boolean
isExisted
(
final
String
key
)
{
return
null
!=
get
(
key
);
}
@Override
public
List
<
String
>
getChildrenKeys
(
final
String
key
)
{
String
path
=
key
+
"/"
;
final
RangeRequest
request
=
RangeRequest
.
newBuilder
().
setKey
(
ByteString
.
copyFromUtf8
(
path
)).
setRangeEnd
(
getRangeEnd
(
path
)).
build
();
Optional
<
List
<
String
>>
result
=
etcdRetryEngine
.
execute
(
new
Callable
<
List
<
String
>>()
{
@Override
public
List
<
String
>
call
()
throws
InterruptedException
,
ExecutionException
,
TimeoutException
{
RangeResponse
response
=
kvStub
.
range
(
request
).
get
(
config
.
getOperationTimeoutMilliseconds
(),
TimeUnit
.
MILLISECONDS
);
List
<
String
>
result
=
new
ArrayList
<>();
for
(
KeyValue
each
:
response
.
getKvsList
())
{
String
childFullPath
=
each
.
getKey
().
toStringUtf8
();
result
.
add
(
childFullPath
.
substring
(
childFullPath
.
lastIndexOf
(
"/"
)
+
1
));
}
return
result
;
}
});
return
result
.
isPresent
()
?
result
.
get
()
:
Collections
.<
String
>
emptyList
();
}
@Override
public
void
persist
(
final
String
key
,
final
String
value
)
{
final
PutRequest
request
=
PutRequest
.
newBuilder
().
setPrevKv
(
true
).
setKey
(
ByteString
.
copyFromUtf8
(
key
)).
setValue
(
ByteString
.
copyFromUtf8
(
value
)).
build
();
etcdRetryEngine
.
execute
(
new
Callable
<
Void
>()
{
@Override
public
Void
call
()
throws
InterruptedException
,
ExecutionException
,
TimeoutException
{
kvStub
.
put
(
request
).
get
(
config
.
getOperationTimeoutMilliseconds
(),
TimeUnit
.
MILLISECONDS
);
return
null
;
}
});
}
@Override
public
void
update
(
final
String
key
,
final
String
value
)
{
persist
(
key
,
value
);
}
@Override
public
void
persistEphemeral
(
final
String
key
,
final
String
value
)
{
final
Optional
<
Long
>
leaseId
=
lease
();
if
(!
leaseId
.
isPresent
())
{
throw
new
RegistryCenterException
(
"Unable to set up heat beat for key %s"
,
key
);
}
final
PutRequest
request
=
PutRequest
.
newBuilder
().
setPrevKv
(
true
).
setLease
(
leaseId
.
get
()).
setKey
(
ByteString
.
copyFromUtf8
(
key
)).
setValue
(
ByteString
.
copyFromUtf8
(
value
)).
build
();
etcdRetryEngine
.
execute
(
new
Callable
<
Void
>()
{
@Override
public
Void
call
()
throws
InterruptedException
,
ExecutionException
,
TimeoutException
{
kvStub
.
put
(
request
).
get
(
config
.
getOperationTimeoutMilliseconds
(),
TimeUnit
.
MILLISECONDS
);
return
null
;
}
});
}
private
Optional
<
Long
>
lease
()
{
final
LeaseGrantRequest
request
=
LeaseGrantRequest
.
newBuilder
().
setTTL
(
config
.
getTimeToLiveSeconds
()).
build
();
return
etcdRetryEngine
.
execute
(
new
Callable
<
Long
>()
{
@Override
public
Long
call
()
throws
InterruptedException
,
ExecutionException
,
TimeoutException
{
long
leaseId
=
leaseStub
.
leaseGrant
(
request
).
get
(
config
.
getOperationTimeoutMilliseconds
(),
TimeUnit
.
MILLISECONDS
).
getID
();
keepAlive
.
heartbeat
(
leaseId
);
return
leaseId
;
}
});
}
@Override
public
void
watch
(
final
String
key
,
final
DataChangedEventListener
dataChangedEventListener
)
{
WatchCreateRequest
createWatchRequest
=
WatchCreateRequest
.
newBuilder
().
setKey
(
ByteString
.
copyFromUtf8
(
key
)).
setRangeEnd
(
getRangeEnd
(
key
)).
build
();
final
WatchRequest
request
=
WatchRequest
.
newBuilder
().
setCreateRequest
(
createWatchRequest
).
build
();
etcdRetryEngine
.
execute
(
new
Callable
<
Void
>()
{
@Override
public
Void
call
()
{
watchStub
.
watch
(
new
EtcdWatchStreamObserver
(
dataChangedEventListener
)).
onNext
(
request
);
return
null
;
}
});
}
@Override
public
void
close
()
{
keepAlive
.
close
();
}
private
ByteString
getRangeEnd
(
final
String
key
)
{
byte
[]
noPrefix
=
{
0
};
byte
[]
endKey
=
key
.
getBytes
().
clone
();
for
(
int
i
=
endKey
.
length
-
1
;
i
>=
0
;
i
--)
{
if
(
endKey
[
i
]
<
0xff
)
{
endKey
[
i
]
=
(
byte
)
(
endKey
[
i
]
+
1
);
return
ByteString
.
copyFrom
(
Arrays
.
copyOf
(
endKey
,
i
+
1
));
}
}
return
ByteString
.
copyFrom
(
noPrefix
);
}
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/channel/EtcdChannelFactory.java
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.shardingsphere.orchestration.reg.etcd.internal.channel
;
import
io.grpc.Channel
;
import
io.grpc.netty.NettyChannelBuilder
;
import
io.grpc.util.RoundRobinLoadBalancerFactory
;
import
lombok.AccessLevel
;
import
lombok.NoArgsConstructor
;
import
java.util.List
;
import
java.util.concurrent.ConcurrentHashMap
;
/**
* Etcd channel factory.
*
* @author zhangliang
*/
@NoArgsConstructor
(
access
=
AccessLevel
.
PRIVATE
)
public
final
class
EtcdChannelFactory
{
private
static
final
String
TARGET
=
"etcd"
;
private
static
ConcurrentHashMap
<
List
<
String
>,
Channel
>
etcdChannels
=
new
ConcurrentHashMap
<>();
/**
* Get etcd channel instance.
*
* @param endpoints etcd endpoints
* @return etcd channel
*/
public
static
Channel
getInstance
(
final
List
<
String
>
endpoints
)
{
if
(
etcdChannels
.
containsKey
(
endpoints
))
{
return
etcdChannels
.
get
(
endpoints
);
}
Channel
channel
=
NettyChannelBuilder
.
forTarget
(
TARGET
)
.
usePlaintext
(
true
)
.
nameResolverFactory
(
new
EtcdNameSolverFactory
(
TARGET
,
endpoints
))
.
loadBalancerFactory
(
RoundRobinLoadBalancerFactory
.
getInstance
())
.
build
();
Channel
result
=
etcdChannels
.
putIfAbsent
(
endpoints
,
channel
);
return
null
==
result
?
channel
:
result
;
}
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/channel/EtcdNameSolverFactory.java
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.shardingsphere.orchestration.reg.etcd.internal.channel
;
import
com.google.common.base.Strings
;
import
com.google.common.collect.Lists
;
import
io.grpc.Attributes
;
import
io.grpc.EquivalentAddressGroup
;
import
io.grpc.NameResolver
;
import
io.grpc.Status
;
import
io.grpc.internal.GrpcUtil
;
import
io.grpc.internal.SharedResourceHolder
;
import
lombok.RequiredArgsConstructor
;
import
lombok.extern.slf4j.Slf4j
;
import
javax.annotation.Nullable
;
import
java.net.InetSocketAddress
;
import
java.net.URI
;
import
java.net.URISyntaxException
;
import
java.util.List
;
import
java.util.concurrent.ExecutorService
;
import
java.util.regex.Pattern
;
/**
* Etcd name solver factory.
*
* @author junxiong
*/
@RequiredArgsConstructor
@Slf4j
public
final
class
EtcdNameSolverFactory
extends
NameResolver
.
Factory
{
private
static
final
Pattern
SCHEMAS
=
Pattern
.
compile
(
"^(http|https)"
);
private
final
String
scheme
;
private
final
List
<
String
>
endpoints
;
private
ExecutorService
executor
;
private
boolean
shutdown
;
@Nullable
@Override
public
NameResolver
newNameResolver
(
final
URI
targetUri
,
final
Attributes
params
)
{
if
(!
scheme
.
equals
(
targetUri
.
getPath
()))
{
return
null
;
}
return
new
NameResolver
()
{
@Override
public
String
getServiceAuthority
()
{
return
scheme
;
}
@Override
public
void
start
(
final
Listener
listener
)
{
if
(
shutdown
)
{
return
;
}
List
<
EquivalentAddressGroup
>
equivalentAddressGroups
=
Lists
.
newArrayList
();
for
(
String
each
:
endpoints
)
{
try
{
URI
uri
=
new
URI
(
each
.
trim
());
if
(!
Strings
.
isNullOrEmpty
(
uri
.
getAuthority
())
&&
SCHEMAS
.
matcher
(
uri
.
getScheme
()).
matches
())
{
InetSocketAddress
inetSocketAddress
=
new
InetSocketAddress
(
uri
.
getHost
(),
uri
.
getPort
());
equivalentAddressGroups
.
add
(
new
EquivalentAddressGroup
(
inetSocketAddress
));
}
}
catch
(
final
URISyntaxException
ex
)
{
listener
.
onError
(
Status
.
INVALID_ARGUMENT
);
log
.
warn
(
"Ignored illegal endpoint, %s"
,
ex
.
getMessage
());
}
}
listener
.
onAddresses
(
equivalentAddressGroups
,
Attributes
.
EMPTY
);
}
@Override
public
void
shutdown
()
{
if
(!
shutdown
)
{
shutdown
=
true
;
executor
=
SharedResourceHolder
.
release
(
GrpcUtil
.
SHARED_CHANNEL_EXECUTOR
,
executor
);
}
}
};
}
@Override
public
String
getDefaultScheme
()
{
return
scheme
;
}
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/keepalive/KeepAlive.java
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.shardingsphere.orchestration.reg.etcd.internal.keepalive
;
import
etcdserverpb.LeaseGrpc
;
import
etcdserverpb.LeaseGrpc.LeaseStub
;
import
etcdserverpb.Rpc.LeaseKeepAliveRequest
;
import
etcdserverpb.Rpc.LeaseKeepAliveResponse
;
import
io.grpc.Channel
;
import
io.grpc.Status
;
import
io.grpc.stub.StreamObserver
;
import
lombok.AllArgsConstructor
;
import
lombok.Setter
;
import
lombok.extern.slf4j.Slf4j
;
import
java.util.concurrent.ConcurrentHashMap
;
import
java.util.concurrent.ConcurrentMap
;
import
java.util.concurrent.Executors
;
import
java.util.concurrent.ScheduledExecutorService
;
import
java.util.concurrent.ScheduledFuture
;
import
java.util.concurrent.TimeUnit
;
/**
* Keep the lease alive.
*
* @author junxiong
*/
@Slf4j
public
final
class
KeepAlive
implements
AutoCloseable
{
private
static
final
long
DELAY_MILLISECONDS
=
100L
;
private
final
LeaseStub
leaseStub
;
private
final
long
heartbeatIntervalMilliseconds
;
private
final
ConcurrentMap
<
Long
,
KeepAliveTask
>
keepAliveTasks
;
private
final
ScheduledFuture
scheduledFuture
;
private
final
ScheduledExecutorService
scheduledService
;
public
KeepAlive
(
final
Channel
channel
,
final
long
timeToLiveSeconds
)
{
leaseStub
=
LeaseGrpc
.
newStub
(
channel
);
heartbeatIntervalMilliseconds
=
timeToLiveSeconds
*
1000L
/
3L
;
keepAliveTasks
=
new
ConcurrentHashMap
<>();
scheduledService
=
Executors
.
newScheduledThreadPool
(
1
);
scheduledFuture
=
scheduledService
.
scheduleAtFixedRate
(
new
Runnable
()
{
@Override
public
void
run
()
{
for
(
KeepAliveTask
keepAliveTask
:
keepAliveTasks
.
values
())
{
keepAliveTask
.
heartbeat
();
}
}
},
DELAY_MILLISECONDS
,
heartbeatIntervalMilliseconds
,
TimeUnit
.
MILLISECONDS
);
}
/**
* keep lease alive.
*
* @param leaseId lease ID
*/
public
void
heartbeat
(
final
long
leaseId
)
{
keepAliveTasks
.
putIfAbsent
(
leaseId
,
new
KeepAliveTask
(
leaseId
,
leaseStub
.
leaseKeepAlive
(
createResponseObserver
(
leaseId
)),
System
.
currentTimeMillis
()));
}
private
StreamObserver
<
LeaseKeepAliveResponse
>
createResponseObserver
(
final
long
leaseId
)
{
return
new
StreamObserver
<
LeaseKeepAliveResponse
>()
{
@Override
public
void
onNext
(
final
LeaseKeepAliveResponse
response
)
{
long
leaseId
=
response
.
getID
();
long
nextHeartbeatTimestamp
=
System
.
currentTimeMillis
()
+
response
.
getTTL
()
*
1000L
/
3L
;
log
.
debug
(
"Reschedule heartbeat time for lease {} to {}"
,
leaseId
,
nextHeartbeatTimestamp
);
KeepAliveTask
keepAliveTask
=
keepAliveTasks
.
get
(
leaseId
);
if
(
null
!=
keepAliveTask
)
{
keepAliveTask
.
setNextHeartbeatTimestamp
(
nextHeartbeatTimestamp
);
}
}
@Override
public
void
onCompleted
()
{
log
.
debug
(
"Keep alive finished"
);
}
@Override
public
void
onError
(
final
Throwable
cause
)
{
log
.
warn
(
"Keep alive failed, due to {}, renew it"
,
Status
.
fromThrowable
(
cause
));
heartbeat
(
leaseId
);
}
};
}
@Override
public
void
close
()
{
for
(
KeepAliveTask
keepAliveTask:
keepAliveTasks
.
values
())
{
keepAliveTask
.
close
();
}
keepAliveTasks
.
clear
();
scheduledService
.
shutdown
();
scheduledFuture
.
cancel
(
false
);
}
@AllArgsConstructor
private
class
KeepAliveTask
implements
AutoCloseable
{
private
final
long
leaseId
;
private
final
StreamObserver
<
LeaseKeepAliveRequest
>
observer
;
@Setter
private
long
nextHeartbeatTimestamp
;
/**
* keep heartbeat.
*/
public
void
heartbeat
()
{
if
(
nextHeartbeatTimestamp
<=
System
.
currentTimeMillis
())
{
log
.
debug
(
"Heartbeat lease {} at time {}"
,
leaseId
,
nextHeartbeatTimestamp
);
observer
.
onNext
(
LeaseKeepAliveRequest
.
newBuilder
().
setID
(
leaseId
).
build
());
}
}
@Override
public
void
close
()
{
observer
.
onCompleted
();
}
}
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/retry/EtcdRetryEngine.java
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.shardingsphere.orchestration.reg.etcd.internal.retry
;
import
com.github.rholder.retry.RetryException
;
import
com.github.rholder.retry.Retryer
;
import
com.github.rholder.retry.RetryerBuilder
;
import
com.github.rholder.retry.StopStrategies
;
import
com.github.rholder.retry.WaitStrategies
;
import
com.google.common.base.Optional
;
import
lombok.RequiredArgsConstructor
;
import
org.apache.shardingsphere.orchestration.reg.api.RegistryCenterConfiguration
;
import
org.apache.shardingsphere.orchestration.reg.exception.RegistryCenterException
;
import
java.util.concurrent.Callable
;
import
java.util.concurrent.ExecutionException
;
import
java.util.concurrent.TimeUnit
;
import
java.util.concurrent.TimeoutException
;
/**
* Etcd retry engine.
*
* @author junxiong
*/
@RequiredArgsConstructor
public
final
class
EtcdRetryEngine
{
private
final
RegistryCenterConfiguration
config
;
/**
* Retry to execute callable command.
*
* @param callable callable command
* @param <T> return type
* @return execute result
*/
public
<
T
>
Optional
<
T
>
execute
(
final
Callable
<
T
>
callable
)
{
Retryer
<
T
>
retryer
=
RetryerBuilder
.<
T
>
newBuilder
()
.
retryIfExceptionOfType
(
TimeoutException
.
class
)
.
retryIfExceptionOfType
(
ExecutionException
.
class
)
.
retryIfExceptionOfType
(
InterruptedException
.
class
)
.
withWaitStrategy
(
WaitStrategies
.
fixedWait
(
config
.
getRetryIntervalMilliseconds
(),
TimeUnit
.
MILLISECONDS
))
.
withStopStrategy
(
StopStrategies
.
stopAfterAttempt
(
config
.
getMaxRetries
()))
.
build
();
try
{
return
Optional
.
fromNullable
(
retryer
.
call
(
callable
));
}
catch
(
final
ExecutionException
|
RetryException
ex
)
{
throw
new
RegistryCenterException
(
ex
);
}
}
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/java/org/apache/shardingsphere/orchestration/reg/etcd/internal/watcher/EtcdWatchStreamObserver.java
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.shardingsphere.orchestration.reg.etcd.internal.watcher
;
import
etcdserverpb.Rpc
;
import
etcdserverpb.Rpc.WatchResponse
;
import
io.grpc.stub.StreamObserver
;
import
lombok.RequiredArgsConstructor
;
import
mvccpb.Kv.Event
;
import
org.apache.shardingsphere.orchestration.reg.exception.RegistryCenterException
;
import
org.apache.shardingsphere.orchestration.reg.listener.DataChangedEvent
;
import
org.apache.shardingsphere.orchestration.reg.listener.DataChangedEvent.ChangedType
;
import
org.apache.shardingsphere.orchestration.reg.listener.DataChangedEventListener
;
/**
* Watch stream observer.
*
* @author junxiong
*/
@RequiredArgsConstructor
public
final
class
EtcdWatchStreamObserver
implements
StreamObserver
<
WatchResponse
>
{
private
final
DataChangedEventListener
dataChangedEventListener
;
@Override
public
void
onNext
(
final
Rpc
.
WatchResponse
response
)
{
if
(
response
.
getCanceled
()
||
response
.
getCreated
())
{
return
;
}
for
(
Event
event
:
response
.
getEventsList
())
{
ChangedType
changedType
=
getChangedType
(
event
);
if
(
ChangedType
.
IGNORED
!=
changedType
)
{
dataChangedEventListener
.
onChange
(
new
DataChangedEvent
(
event
.
getKv
().
getKey
().
toStringUtf8
(),
event
.
getKv
().
getValue
().
toStringUtf8
(),
changedType
));
}
}
}
private
ChangedType
getChangedType
(
final
Event
event
)
{
switch
(
event
.
getType
())
{
case
PUT:
return
DataChangedEvent
.
ChangedType
.
UPDATED
;
case
DELETE:
return
DataChangedEvent
.
ChangedType
.
DELETED
;
default
:
return
DataChangedEvent
.
ChangedType
.
IGNORED
;
}
}
@Override
public
void
onError
(
final
Throwable
throwable
)
{
// TODO retry watch later
throw
new
RegistryCenterException
(
new
Exception
(
throwable
));
}
@Override
public
void
onCompleted
()
{
}
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/proto/etcd/auth/authpb/auth.proto
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.
*/
syntax
=
"proto3"
;
package
authpb
;
// User is a single entry in the bucket authUsers
message
User
{
bytes
name
=
1
;
bytes
password
=
2
;
repeated
string
roles
=
3
;
}
// Permission is a single entity
message
Permission
{
enum
Type
{
READ
=
0
;
WRITE
=
1
;
READWRITE
=
2
;
}
Type
permType
=
1
;
bytes
key
=
2
;
bytes
range_end
=
3
;
}
// Role is a single entry in the bucket authRoles
message
Role
{
bytes
name
=
1
;
repeated
Permission
keyPermission
=
2
;
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/proto/etcd/etcdserver/etcdserverpb/rpc.proto
已删除
100644 → 0
浏览文件 @
da082d2a
此差异已折叠。
点击以展开。
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/proto/etcd/mvcc/mvccpb/kv.proto
已删除
100644 → 0
浏览文件 @
da082d2a
/*
* 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.
*/
syntax
=
"proto3"
;
package
mvccpb
;
message
KeyValue
{
// key is the key in bytes. An empty key is not allowed.
bytes
key
=
1
;
// create_revision is the revision of last creation on this key.
int64
create_revision
=
2
;
// mod_revision is the revision of last modification on this key.
int64
mod_revision
=
3
;
// version is the version of the key. A deletion resets
// the version to zero and any modification of the key
// increases its version.
int64
version
=
4
;
// value is the value held by the key, in bytes.
bytes
value
=
5
;
// lease is the ID of the lease that attached to key.
// When the attached lease expires, the key will be deleted.
// If lease is 0, then no lease is attached to the key.
int64
lease
=
6
;
}
message
Event
{
enum
EventType
{
PUT
=
0
;
DELETE
=
1
;
}
// type is the kind of event. If type is a PUT, it indicates
// new data has been stored to the key. If type is a DELETE,
// it indicates the key was deleted.
EventType
type
=
1
;
// kv holds the KeyValue for the event.
// A PUT event contains current kv pair.
// A PUT event with kv.Version=1 indicates the creation of a key.
// A DELETE/EXPIRE event contains the deleted key with
// its modification revision set to the revision of deletion.
KeyValue
kv
=
2
;
// prev_kv holds the key-value pair before the event happens.
KeyValue
prev_kv
=
3
;
}
sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.orchestration.reg.api.RegistryCenter
已删除
100644 → 0
浏览文件 @
da082d2a
#
# 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.
#
org.apache.shardingsphere.orchestration.reg.etcd.EtcdRegistryCenter
shardingsphere-distribution/shardingsphere-basic-distribution/pom.xml
浏览文件 @
689dce33
...
...
@@ -189,11 +189,6 @@
<artifactId>
sharding-orchestration-reg-zookeeper-curator
</artifactId>
<version>
${project.version}
</version>
</dependency>
<dependency>
<groupId>
org.apache.shardingsphere
</groupId>
<artifactId>
sharding-orchestration-reg-etcd
</artifactId>
<version>
${project.version}
</version>
</dependency>
<dependency>
<groupId>
org.apache.shardingsphere
</groupId>
<artifactId>
sharding-opentracing
</artifactId>
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录