This repository has been archived by the owner on Mar 31, 2023. It is now read-only.
Send GoalStateV2 from DPM, NCM and ACA #704
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -53,6 +53,8 @@ spec: | |
| replicas: 1 | ||
| template: | ||
| metadata: | ||
| annotations: | ||
| linkerd.io/inject: enabled | ||
| labels: | ||
| app: netwconfigmanager | ||
| spec: | ||
|
|
@@ -69,7 +71,8 @@ spec: | |
| imagePullPolicy: IfNotPresent | ||
| command: ["java", "-jar", "/app/AlcorNetworkConfigManager-0.1.0.jar", "--spring.config.location=/etc/ncm/application.properties"] | ||
| ports: | ||
| - containerPort: 50001 | ||
| protocol: TCP | ||
| volumeMounts: | ||
| - name: ncm-volume | ||
| mountPath: /etc/ncm | ||
|
|
@@ -84,10 +87,10 @@ metadata: | |
| labels: | ||
| name: netwconfigmanager-service | ||
| spec: | ||
| type: ClusterIP | ||
| ports: | ||
| - port: 50001 | ||
| protocol: TCP | ||
| targetPort: 50001 | ||
|
Comment on lines
+92
to
+94
There was a problem hiding this comment. @DavidLiu506 Do we need to include port spec for restful here too? There was a problem hiding this comment. Yes, we can add restful configure after restful api fix.
Comment on lines
+92
to
+94
There was a problem hiding this comment. Would this change block port 9014 for REST APIs? There was a problem hiding this comment. This is for grpc configure, we can add another service for restful. |
||
| selector: | ||
| app: netwconfigmanager | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,9 +22,7 @@ | |
| import com.futurewei.alcor.dataplane.config.Config; | ||
| import com.futurewei.alcor.dataplane.entity.MulticastGoalStateV2; | ||
| import com.futurewei.alcor.dataplane.entity.UnicastGoalStateV2; | ||
| import com.futurewei.alcor.schema.*; | ||
| import io.grpc.ConnectivityState; | ||
| import io.grpc.ManagedChannel; | ||
| import io.grpc.ManagedChannelBuilder; | ||
|
|
@@ -37,6 +35,8 @@ | |
|
|
||
| import java.util.*; | ||
| import java.util.concurrent.*; | ||
| import java.util.logging.Level; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| @Service("grpcDataPlaneClient") | ||
| @ConditionalOnProperty(prefix = "protobuf.goal-state-message", name = "version", havingValue = "102") | ||
|
|
@@ -56,21 +56,27 @@ public class DataPlaneClientImplV2 implements DataPlaneClient<UnicastGoalStateV2 | |
| // when a channel is set up, send this amount of default GoalStates for warmup. | ||
| private int numberOfWarmupsPerChannel; | ||
|
|
||
| private String netwconfigmanagerHost; | ||
|
|
||
| // prints out UUID and time, when sending a GoalState to any of the monitorHosts | ||
| private ArrayList<String> monitorHosts; | ||
|
|
||
| private ConcurrentHashMap<String, ArrayList<GrpcChannelStub>> hostIpGrpcChannelStubMap; | ||
|
|
||
| @Override | ||
| public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates) throws Exception { | ||
| Goalstate.GoalStateV2.Builder goalStateBuilder = Goalstate.GoalStateV2.newBuilder(); | ||
| final CountDownLatch finishLatch = new CountDownLatch(1); | ||
| List<String> results = new ArrayList<>(); | ||
| for (UnicastGoalStateV2 unicastGoalState : unicastGoalStates) { | ||
| goalStateBuilder = getGoalState(goalStateBuilder, unicastGoalState); | ||
| } | ||
| doSendGoalState(goalStateBuilder.build(), finishLatch, results); | ||
|
|
||
| if (!finishLatch.await(1, TimeUnit.MINUTES)) { | ||
| LOG.warn("Send goal states can not finish within 1 minutes"); | ||
| return Arrays.asList("Send goal states can not finish within 1 minutes"); | ||
| } | ||
|
Comment on lines
+76
to
+79
There was a problem hiding this comment. @DavidLiu506 Do we know which goalstate or which ncm cannot sent to? If we do, can we put the id or uri in the log and return results? There was a problem hiding this comment. BTW, can we put 1 mins as a variable in application.properties? There was a problem hiding this comment. NCM managed by linked, there only one goal state v2 send to NCM. There was a problem hiding this comment. Talked with James, NCM already have ACA response information. DPM don't need to know it. |
||
| return results; | ||
| } | ||
|
|
||
|
|
@@ -92,6 +98,10 @@ public DataPlaneClientImplV2(Config globalConfig, ArrayList<String> monitorHosts | |
| numberOfWarmupsPerChannel = 0; | ||
| } | ||
|
|
||
| if (netwconfigmanagerHost == null || netwconfigmanagerHost.isEmpty()) { | ||
| netwconfigmanagerHost = globalConfig.netwconfigmanagerHost; | ||
| } | ||
|
|
||
| this.monitorHosts = monitorHosts; | ||
| LOG.info("Printing out all monitorHosts"); | ||
| for(String host : this.monitorHosts){ | ||
|
|
@@ -100,7 +110,7 @@ public DataPlaneClientImplV2(Config globalConfig, ArrayList<String> monitorHosts | |
| LOG.info("Done printing out all monitorHosts"); | ||
| this.numberOfGrpcChannelPerHost = globalConfig.numberOfGrpcChannelPerHost; | ||
| this.numberOfWarmupsPerChannel = globalConfig.numberOfWarmupsPerChannel; | ||
| this.hostAgentPort = 9016; | ||
xieus marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| this.executor = new ThreadPoolExecutor(100, | ||
| 200, | ||
|
|
@@ -173,7 +183,8 @@ private ArrayList<GrpcChannelStub> createGrpcChannelStubArrayList(String hostIp) | |
| ArrayList<GrpcChannelStub> arr = new ArrayList<>(); | ||
| for (int i = 0; i < numberOfGrpcChannelPerHost; i++) { | ||
| GrpcChannelStub channelStub = createGrpcChannelStub(hostIp); | ||
| // Using Linkerd load balance | ||
| //warmUpChannelStub(channelStub, hostIp); | ||
xieus marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| arr.add(channelStub); | ||
| } | ||
| long end = System.currentTimeMillis(); | ||
|
|
@@ -220,11 +231,9 @@ public void onCompleted() { | |
| return; | ||
| } | ||
|
|
||
| private String doSendGoalState(Goalstate.GoalStateV2 goalStateV2, CountDownLatch finishLatch, List<String> replies) { | ||
| String hostIp = netwconfigmanagerHost; | ||
| long start = System.currentTimeMillis(); | ||
| GrpcChannelStub channelStub = getOrCreateGrpcChannel(hostIp); | ||
| long chan_established = System.currentTimeMillis(); | ||
| LOG.info("[doSendGoalState] Established channel, elapsed Time in milli seconds: " + (chan_established - start)); | ||
|
|
@@ -239,6 +248,12 @@ private String doSendGoalState(UnicastGoalStateV2 unicastGoalState) { | |
| public void onNext(Goalstateprovisioner.GoalStateOperationReply reply) { | ||
| LOG.info("Receive response from ACA@" + hostIp + " | " + reply.toString()); | ||
| result.put(hostIp, reply.getOperationStatusesList()); | ||
| if (reply.getOperationStatusesList().stream().filter(item -> item.getOperationStatus().equals(Common.OperationStatus.FAILURE)).collect(Collectors.toList()).size() > 0) { | ||
| replies.add(reply.toString()); | ||
| while (finishLatch.getCount() > 0) { | ||
| finishLatch.countDown(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -249,22 +264,13 @@ public void onError(Throwable t) { | |
| @Override | ||
| public void onCompleted() { | ||
| LOG.info("Complete receiving message from ACA@" + hostIp); | ||
| finishLatch.countDown(); | ||
| } | ||
| }; | ||
|
|
||
| StreamObserver<Goalstate.GoalStateV2> requestObserver = asyncStub.pushGoalStatesStream(responseObserver); | ||
| try { | ||
| requestObserver.onNext(goalStateV2); | ||
| } catch (RuntimeException e) { | ||
| // Cancel RPC | ||
| LOG.warn("[doSendGoalState] Sending GS, but error happened | " + e.getMessage()); | ||
|
|
@@ -275,10 +281,7 @@ public void onCompleted() { | |
| LOG.info("Sending GS to Host " + hostIp + " is completed"); | ||
|
|
||
| // comment out onCompleted so that the same channel/stub and keep sending next time. | ||
| requestObserver.onCompleted(); | ||
| return null; | ||
| } | ||
|
|
||
|
|
@@ -291,4 +294,115 @@ public GrpcChannelStub(ManagedChannel channel, GoalStateProvisionerGrpc.GoalStat | |
| this.stub = stub; | ||
| } | ||
| } | ||
|
|
||
| private Goalstate.GoalStateV2.Builder getGoalState(Goalstate.GoalStateV2.Builder goalStateBuilder, UnicastGoalStateV2 unicastGoalStateV2) { | ||
| Goalstate.GoalStateV2 goalStateV2 = unicastGoalStateV2.getGoalState(); | ||
| Goalstate.HostResources.Builder hostResourceBuilder = Goalstate.HostResources.newBuilder(); | ||
| if (goalStateV2.getSubnetStatesCount() > 0) { | ||
| goalStateV2.getSubnetStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType subnetResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.SUBNET) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(subnetResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllSubnetStates(goalStateV2.getSubnetStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateV2.getDhcpStatesCount() > 0) { | ||
| goalStateV2.getDhcpStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType dhcpResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.DHCP) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(dhcpResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllDhcpStates(goalStateV2.getDhcpStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateV2.getPortStatesCount() > 0) { | ||
| goalStateV2.getPortStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType portResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.PORT) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(portResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllPortStates(goalStateV2.getPortStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateV2.getSecurityGroupStatesCount() > 0) { | ||
| goalStateV2.getSecurityGroupStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType securityGroupResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.SECURITYGROUP) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(securityGroupResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllSecurityGroupStates(goalStateV2.getSecurityGroupStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateV2.getNeighborStatesCount() > 0) { | ||
| goalStateV2.getNeighborStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType neighborGroupResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.NEIGHBOR) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(neighborGroupResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllNeighborStates(goalStateV2.getNeighborStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateV2.getRouterStatesCount() > 0) { | ||
| goalStateV2.getRouterStatesMap().entrySet().forEach(entry -> { | ||
| Goalstate.ResourceIdType routerResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.ROUTER) | ||
| .setId(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()) | ||
| .build(); | ||
| hostResourceBuilder.addResources(routerResourceIdType); | ||
|
|
||
|
|
||
| if (goalStateBuilder.containsRouterStates(unicastGoalStateV2.getHostIp() + "/" + entry.getKey())) { | ||
| Router.RouterConfiguration.Builder routerConfigurationBuilder = goalStateBuilder.getRouterStatesMap().get(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()).getConfiguration().toBuilder(); | ||
| routerConfigurationBuilder.addAllSubnetRoutingTables(entry.getValue().getConfiguration().getSubnetRoutingTablesList()); | ||
| Router.RouterState.Builder routerStateBuilder = goalStateBuilder.getRouterStatesMap().get(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()).toBuilder(); | ||
| routerStateBuilder.setConfiguration(routerConfigurationBuilder); | ||
| } else { | ||
| goalStateBuilder.putRouterStates(unicastGoalStateV2.getHostIp() + "/" + entry.getKey(), entry.getValue()); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| if (goalStateV2.getVpcStatesCount() > 0) { | ||
| goalStateV2.getVpcStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType vpcResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.VPC) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(vpcResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllVpcStates(goalStateV2.getVpcStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateV2.getGatewayStatesCount() > 0) { | ||
| goalStateV2.getGatewayStatesMap().keySet().forEach(key -> { | ||
| Goalstate.ResourceIdType gatewayResourceIdType = Goalstate.ResourceIdType.newBuilder() | ||
| .setType(Common.ResourceType.GATEWAY) | ||
| .setId(key) | ||
| .build(); | ||
| hostResourceBuilder.addResources(gatewayResourceIdType); | ||
| }); | ||
| goalStateBuilder.putAllGatewayStates(goalStateV2.getGatewayStatesMap()); | ||
| } | ||
|
|
||
| if (goalStateBuilder.containsHostResources(unicastGoalStateV2.getHostIp())) { | ||
| Goalstate.HostResources.Builder hostResourceBuilder1 = Goalstate.HostResources.newBuilder(); | ||
| hostResourceBuilder1.addAllResources(hostResourceBuilder.getResourcesList()); | ||
| hostResourceBuilder1.addAllResources(goalStateBuilder.getHostResourcesMap().get(unicastGoalStateV2.getHostIp()).getResourcesList()); | ||
| goalStateBuilder.putHostResources(unicastGoalStateV2.getHostIp(), hostResourceBuilder1.build()); | ||
| } else { | ||
| goalStateBuilder.putHostResources(unicastGoalStateV2.getHostIp(), hostResourceBuilder.build()); | ||
| } | ||
| return goalStateBuilder; | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@DavidLiu506 Why we need to change 8080 to 50001 here? Is this for grpc connection?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, this is for grpc connection.