Skip to content

Commit cd003d5

Browse files
awang12345Nicole00jiangyiwang-jk
authored
Add storage address mapping transform (#589)
* fix conflict * Add storage address mapping transform * update --------- Co-authored-by: Anqi <anqi.wang@vesoft.com> Co-authored-by: jiangyiwang-jk <jiangyiwang-jk@360shuke.com>
1 parent 6b25e64 commit cd003d5

File tree

5 files changed

+219
-19
lines changed

5 files changed

+219
-19
lines changed

client/src/main/java/com/vesoft/nebula/client/meta/MetaClient.java

-1
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@
1010
import com.facebook.thrift.transport.THeaderTransport;
1111
import com.facebook.thrift.transport.TSocket;
1212
import com.facebook.thrift.transport.TTransportException;
13-
import com.google.common.base.Charsets;
1413
import com.vesoft.nebula.ErrorCode;
1514
import com.vesoft.nebula.HostAddr;
1615
import com.vesoft.nebula.client.graph.data.CASignedSSLParam;

client/src/main/java/com/vesoft/nebula/client/meta/MetaManager.java

+51-2
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import com.vesoft.nebula.meta.IdName;
1717
import com.vesoft.nebula.meta.SpaceItem;
1818
import com.vesoft.nebula.meta.TagItem;
19+
import com.vesoft.nebula.util.NetUtil;
1920
import java.io.Serializable;
2021
import java.net.UnknownHostException;
2122
import java.util.ArrayList;
@@ -24,7 +25,9 @@
2425
import java.util.List;
2526
import java.util.Map;
2627
import java.util.Set;
28+
import java.util.concurrent.ConcurrentHashMap;
2729
import java.util.concurrent.locks.ReentrantReadWriteLock;
30+
import java.util.stream.Collectors;
2831
import org.slf4j.Logger;
2932
import org.slf4j.LoggerFactory;
3033

@@ -44,6 +47,8 @@ private class SpaceInfo {
4447
private Map<String, MetaManager.SpaceInfo> spacesInfo = new HashMap<>();
4548
private Map<String, Map<Integer, HostAddr>> partLeaders = null;
4649

50+
private Map<HostAddr, HostAddr> storageAddressMapping = new ConcurrentHashMap<>();
51+
4752
private static final Logger LOGGER = LoggerFactory.getLogger(MetaManager.class);
4853

4954
private MetaClient metaClient;
@@ -75,6 +80,33 @@ public MetaManager(List<HostAddress> address, int timeout, int connectionRetry,
7580
fillMetaInfo();
7681
}
7782

83+
/**
84+
* Add address mapping for storage.Used for change address of storage read from meta server.
85+
*
86+
* @param sourceAddr ip:port
87+
* @param targetAddr ip:port
88+
*/
89+
public void addStorageAddrMapping(String sourceAddr, String targetAddr) {
90+
if (sourceAddr != null && targetAddr != null) {
91+
storageAddressMapping.put(NetUtil.parseHostAddr(sourceAddr),
92+
NetUtil.parseHostAddr(targetAddr));
93+
}
94+
}
95+
96+
/**
97+
* Add address mapping for storage.Used for change address of storage read from meta server.
98+
*
99+
* @param addressMap sourceAddr(ip:port) => targetAddr(ip:port)
100+
*/
101+
public void addStorageAddrMapping(Map<String, String> addressMap) {
102+
if (addressMap != null && !addressMap.isEmpty()) {
103+
for (Map.Entry<String, String> et : addressMap.entrySet()) {
104+
storageAddressMapping.put(NetUtil.parseHostAddr(et.getKey()),
105+
NetUtil.parseHostAddr(et.getValue()));
106+
}
107+
}
108+
}
109+
78110

79111
/**
80112
* close meta client
@@ -280,7 +312,8 @@ public HostAddr getLeader(String spaceName, int part) throws IllegalArgumentExce
280312
if (!partLeaders.get(spaceName).containsKey(part)) {
281313
throw new IllegalArgumentException("PartId:" + part + " does not exist.");
282314
}
283-
return partLeaders.get(spaceName).get(part);
315+
HostAddr hostAddr = partLeaders.get(spaceName).get(part);
316+
return storageAddressMapping.getOrDefault(hostAddr, hostAddr);
284317
} finally {
285318
lock.readLock().unlock();
286319
}
@@ -313,7 +346,17 @@ public Map<Integer, List<HostAddr>> getPartsAlloc(String spaceName)
313346
if (!spacesInfo.containsKey(spaceName)) {
314347
throw new IllegalArgumentException("Space:" + spaceName + " does not exist.");
315348
}
316-
return spacesInfo.get(spaceName).partsAlloc;
349+
Map<Integer, List<HostAddr>> partsAlloc = spacesInfo.get(spaceName).partsAlloc;
350+
if (!storageAddressMapping.isEmpty()) {
351+
// transform real address to special address by mapping
352+
partsAlloc.keySet().forEach(partId -> {
353+
partsAlloc.computeIfPresent(partId, (k, addressList) -> addressList
354+
.stream()
355+
.map(hostAddr -> storageAddressMapping.getOrDefault(hostAddr, hostAddr))
356+
.collect(Collectors.toList()));
357+
});
358+
}
359+
return partsAlloc;
317360
} finally {
318361
lock.readLock().unlock();
319362
}
@@ -355,6 +398,12 @@ public Set<HostAddr> listHosts() {
355398
if (hosts == null) {
356399
return new HashSet<>();
357400
}
401+
if (!storageAddressMapping.isEmpty()) {
402+
hosts = hosts
403+
.stream()
404+
.map(hostAddr -> storageAddressMapping.getOrDefault(hostAddr, hostAddr))
405+
.collect(Collectors.toSet());
406+
}
358407
return hosts;
359408
}
360409

client/src/main/java/com/vesoft/nebula/client/storage/StorageClient.java

+20-16
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import java.util.Arrays;
2424
import java.util.HashSet;
2525
import java.util.List;
26+
import java.util.Map;
2627
import java.util.Set;
2728
import org.slf4j.Logger;
2829
import org.slf4j.LoggerFactory;
@@ -41,8 +42,7 @@ public class StorageClient implements Serializable {
4142
private boolean enableSSL = false;
4243
private SSLParam sslParam = null;
4344

44-
private String user = null;
45-
private String password = null;
45+
private Map<String, String> storageAddressMapping;
4646

4747
/**
4848
* Get a Nebula Storage client that executes the scan query to get NebulaGraph's data with
@@ -95,6 +95,23 @@ public StorageClient(List<HostAddress> addresses, int timeout, int connectionRet
9595
}
9696
}
9797

98+
/**
99+
* The storage address translation relationship is set to convert the storage address
100+
* that cannot be obtained by requesting the meta service
101+
*
102+
* @param storageAddressMapping sourceAddressFromMeta -> targetAddress,Format ip:port.
103+
* eg: 127.0.0.1:9559 -> 10.1.1.2:9559,
104+
* Translates the storage 127.0.0.1:9559 address obtained from the
105+
* meta server to 10.1.1.2:9559. It will use 10.1.1.2:9559 to
106+
* request storage. Instead of 27.0.0.1:9559
107+
*/
108+
public void setStorageAddressMapping(Map<String, String> storageAddressMapping) {
109+
this.storageAddressMapping = storageAddressMapping;
110+
if (this.metaManager != null) {
111+
this.metaManager.addStorageAddrMapping(storageAddressMapping);
112+
}
113+
}
114+
98115
/**
99116
* Connect to Nebula Storage server.
100117
*
@@ -108,19 +125,10 @@ public boolean connect() throws Exception {
108125
pool = new StorageConnPool(config);
109126
metaManager = new MetaManager(addresses, timeout, connectionRetry, executionRetry,
110127
enableSSL, sslParam);
128+
metaManager.addStorageAddrMapping(storageAddressMapping);
111129
return true;
112130
}
113131

114-
public StorageClient setUser(String user) {
115-
this.user = user;
116-
return this;
117-
}
118-
119-
public StorageClient setPassword(String password) {
120-
this.password = password;
121-
return this;
122-
}
123-
124132

125133
/**
126134
* scan vertex of all parts with specific return cols, if returnCols is an empty list, then
@@ -632,8 +640,6 @@ private ScanVertexResultIterator doScanVertex(String spaceName,
632640
.withSpaceName(spaceName)
633641
.withTagName(tagName)
634642
.withPartSuccess(allowPartSuccess)
635-
.withUser(user)
636-
.withPassword(password)
637643
.build();
638644
}
639645

@@ -1096,8 +1102,6 @@ private ScanEdgeResultIterator doScanEdge(String spaceName,
10961102
.withSpaceName(spaceName)
10971103
.withEdgeName(edgeName)
10981104
.withPartSuccess(allowPartSuccess)
1099-
.withUser(user)
1100-
.withPassword(password)
11011105
.build();
11021106
}
11031107

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
package com.vesoft.nebula.util;
2+
3+
import com.vesoft.nebula.HostAddr;
4+
import com.vesoft.nebula.client.graph.data.HostAddress;
5+
6+
/**
7+
* The util of network
8+
*
9+
* @Author jiangyiwang-jk
10+
* @Date 2024/2/1 15:36
11+
*/
12+
public class NetUtil {
13+
14+
private NetUtil() {
15+
;
16+
}
17+
18+
public static HostAddr parseHostAddr(String hostAddress) {
19+
assert hostAddress != null : "Host address should not be null";
20+
String[] hostPort = hostAddress.split(":");
21+
assert hostPort.length == 2 : String.format("Invalid host address %s", hostAddress);
22+
return new HostAddr(hostPort[0].trim(), Integer.parseInt(hostPort[1].trim()));
23+
}
24+
25+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
/* Copyright (c) 2020 vesoft inc. All rights reserved.
2+
*
3+
* This source code is licensed under Apache 2.0 License.
4+
*/
5+
6+
package com.vesoft.nebula.examples;
7+
8+
import com.vesoft.nebula.client.storage.StorageClient;
9+
import com.vesoft.nebula.client.storage.data.EdgeTableRow;
10+
import com.vesoft.nebula.client.storage.data.VertexRow;
11+
import com.vesoft.nebula.client.storage.data.VertexTableRow;
12+
import com.vesoft.nebula.client.storage.scan.ScanEdgeResult;
13+
import com.vesoft.nebula.client.storage.scan.ScanEdgeResultIterator;
14+
import com.vesoft.nebula.client.storage.scan.ScanVertexResult;
15+
import com.vesoft.nebula.client.storage.scan.ScanVertexResultIterator;
16+
import java.util.Arrays;
17+
import java.util.HashMap;
18+
import java.util.List;
19+
import java.util.Map;
20+
import org.slf4j.Logger;
21+
import org.slf4j.LoggerFactory;
22+
23+
public class SpecialAddressStorageClientExample {
24+
private static final Logger LOGGER =
25+
LoggerFactory.getLogger(SpecialAddressStorageClientExample.class);
26+
27+
public static void main(String[] args) {
28+
Map<String,String> storageAddressMapping = new HashMap<>();
29+
storageAddressMapping.put("127.0.0.1:9559","10.xx.xx.xx:9559");
30+
// input params are the metad's ip and port
31+
StorageClient client = new StorageClient("127.0.0.1", 9559);
32+
// set storage address mapping
33+
client.setStorageAddressMapping(storageAddressMapping);
34+
try {
35+
client.connect();
36+
} catch (Exception e) {
37+
LOGGER.error("storage client connect error, ", e);
38+
client.close();
39+
System.exit(1);
40+
}
41+
scanVertex(client);
42+
scanEdge(client);
43+
44+
client.close();
45+
}
46+
47+
/**
48+
* Vertex Person's property in Nebula Graph:
49+
* first_name, last_name, gender, birthday
50+
* Tom Li 男 2010
51+
*/
52+
public static void scanVertex(StorageClient client) {
53+
ScanVertexResultIterator iterator = client.scanVertex(
54+
"test",
55+
"person",
56+
Arrays.asList("name", "age"));
57+
58+
while (iterator.hasNext()) {
59+
ScanVertexResult result = null;
60+
try {
61+
result = iterator.next();
62+
} catch (Exception e) {
63+
LOGGER.error("scan error, ", e);
64+
client.close();
65+
System.exit(1);
66+
}
67+
if (result.isEmpty()) {
68+
continue;
69+
}
70+
71+
List<VertexRow> vertexRows = result.getVertices();
72+
for (VertexRow row : vertexRows) {
73+
if (result.getVertex(row.getVid()) != null) {
74+
System.out.println(result.getVertex(row.getVid()));
75+
}
76+
}
77+
78+
System.out.println("\nresult vertex table view:");
79+
System.out.println(result.getPropNames());
80+
List<VertexTableRow> vertexTableRows = result.getVertexTableRows();
81+
for (VertexTableRow vertex : vertexTableRows) {
82+
System.out.println(vertex.getValues());
83+
}
84+
System.out.println("\n");
85+
}
86+
}
87+
88+
/**
89+
* Edge Friend's property in Nebula Graph:
90+
* degree
91+
* 1.0
92+
*/
93+
public static void scanEdge(StorageClient client) {
94+
ScanEdgeResultIterator iterator = client.scanEdge(
95+
"test",
96+
"like",
97+
Arrays.asList("likeness"));
98+
99+
while (iterator.hasNext()) {
100+
ScanEdgeResult result = null;
101+
try {
102+
result = iterator.next();
103+
} catch (Exception e) {
104+
LOGGER.error("scan error, ", e);
105+
client.close();
106+
System.exit(1);
107+
}
108+
if (result.isEmpty()) {
109+
continue;
110+
}
111+
112+
System.out.println(result.getEdges());
113+
114+
System.out.println("\nresult edge table view:");
115+
System.out.println(result.getPropNames());
116+
List<EdgeTableRow> edgeTableRows = result.getEdgeTableRows();
117+
for (EdgeTableRow edge : edgeTableRows) {
118+
System.out.println(edge.getValues());
119+
}
120+
System.out.println("\n");
121+
}
122+
}
123+
}

0 commit comments

Comments
 (0)