add support for sequence numner in the VPC topology updates and VPC

routing policy updates

Conflicts:
	setup/db/db/schema-430to440.sql

plugins/hypervisors/xen/src/com/cloud/hypervisor/xen/resource/CitrixResourceBase.java

@@ -5306,8 +5306,10 @@ public abstract class CitrixResourceBase implements ServerResource, HypervisorRe
         try {
             Network nw = findOrCreateTunnelNetwork(conn, cmd.getBridgeName());
             String bridgeName = nw.getBridge(conn);
+            long sequenceNo = cmd.getSequenceNumber();
             String result = callHostPlugin(conn, "ovstunnel", "configure_ovs_bridge_for_network_topology", "bridge",
-                    bridgeName, "config", cmd.getVpcConfigInJson(), "host-id", ((Long)cmd.getHostId()).toString());
+                    bridgeName, "config", cmd.getVpcConfigInJson(), "host-id", ((Long)cmd.getHostId()).toString(),
+                    "seq-no", Long.toString(sequenceNo));
                 if (result.startsWith("SUCCESS")) {
                 return new Answer(cmd, true, result);
             } else {
@@ -5324,10 +5326,11 @@ public abstract class CitrixResourceBase implements ServerResource, HypervisorRe
         try {
             Network nw = findOrCreateTunnelNetwork(conn, cmd.getBridgeName());
             String bridgeName = nw.getBridge(conn);
+            long sequenceNo = cmd.getSequenceNumber();
 
             String result = callHostPlugin(conn, "ovstunnel", "configure_ovs_bridge_for_routing_policies", "bridge",
                     bridgeName, "host-id", ((Long)cmd.getHostId()).toString(), "config",
-                    cmd.getVpcConfigInJson());
+                    cmd.getVpcConfigInJson(), "seq-no", Long.toString(sequenceNo));
             if (result.startsWith("SUCCESS")) {
                 return new Answer(cmd, true, result);
             } else {

plugins/network-elements/ovs/resources/META-INF/cloudstack/ovs/spring-ovs-context.xml

@@ -39,4 +39,5 @@
     <bean id="ovsTunnelInterfaceDaoImpl" class="com.cloud.network.ovs.dao.OvsTunnelInterfaceDaoImpl" />
     <bean id="ovsTunnelNetworkDaoImpl" class="com.cloud.network.ovs.dao.OvsTunnelNetworkDaoImpl" />
     <bean id="ovsNetworkTopologyGuruImpl" class="com.cloud.network.ovs.OvsNetworkTopologyGuruImpl"/>
+    <bean id="VpcDistributedRouterSeqNoDaoImpl" class="com.cloud.network.ovs.dao.VpcDistributedRouterSeqNoDaoImpl"/>
   </beans>

plugins/network-elements/ovs/src/com/cloud/agent/api/OvsVpcPhysicalTopologyConfigCommand.java

@@ -32,6 +32,9 @@ public class OvsVpcPhysicalTopologyConfigCommand extends Command {
     long hostId;
     String bridgeName;
 
+    long sequenceNumber;
+    String schemaVersion;
+
     public static class Host {
         long hostId;
         String ipAddress;
@@ -127,4 +130,13 @@ public class OvsVpcPhysicalTopologyConfigCommand extends Command {
     public void setBridgeName(String bridgeName) {
         this.bridgeName = bridgeName;
     }
+
+    public long getSequenceNumber() {
+        return sequenceNumber;
+    }
+
+    public void setSequenceNumber(long sequenceNumber) {
+        this.sequenceNumber = sequenceNumber;
+    }
+
 }

plugins/network-elements/ovs/src/com/cloud/agent/api/OvsVpcRoutingPolicyConfigCommand.java

@@ -27,6 +27,8 @@ public class OvsVpcRoutingPolicyConfigCommand extends Command {
     VpcConfig vpcConfig =null;
     long hostId;
     String bridgeName;
+    long sequenceNumber;
+    String schemaVersion;
 
     public static class AclItem {
         int number;
@@ -120,4 +122,12 @@ public class OvsVpcRoutingPolicyConfigCommand extends Command {
     public boolean executeInSequence() {
         return false;
     }
+
+    public long getSequenceNumber() {
+        return sequenceNumber;
+    }
+
+    public void setSequenceNumber(long sequenceNumber) {
+        this.sequenceNumber = sequenceNumber;
+    }
 }

plugins/network-elements/ovs/src/com/cloud/network/ovs/OvsTunnelManagerImpl.java

@@ -78,9 +78,14 @@ import com.cloud.network.ovs.dao.OvsTunnelInterfaceVO;
 import com.cloud.network.ovs.dao.OvsTunnelNetworkDao;
 import com.cloud.network.ovs.dao.OvsTunnelNetworkVO;
 import com.cloud.network.ovs.dao.OvsTunnel;
+import com.cloud.network.ovs.dao.VpcDistributedRouterSeqNoDao;
+import com.cloud.network.ovs.dao.VpcDistributedRouterSeqNoVO;
 import com.cloud.utils.component.ManagerBase;
 import com.cloud.utils.concurrency.NamedThreadFactory;
 import com.cloud.utils.db.DB;
+import com.cloud.utils.db.TransactionStatus;
+import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionCallback;
 import com.cloud.utils.fsm.StateListener;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.vm.dao.DomainRouterDao;
@@ -128,6 +133,8 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
     NetworkACLDao _networkACLDao;
     @Inject
     NetworkACLItemDao _networkACLItemDao;
+    @Inject
+    VpcDistributedRouterSeqNoDao _vpcDrSeqNoDao;
 
     @Override
     public boolean configure(String name, Map<String, Object> params)
@@ -683,7 +690,7 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
         return true;
     }
 
-    public void handleVmStateChange(VMInstanceVO vm) {
+    private void handleVmStateChange(VMInstanceVO vm) {
 
         // get the VPC's impacted with the VM start
         List<Long> vpcIds = _ovsNetworkToplogyGuru.getVpcIdsVmIsPartOf(vm.getId());
@@ -693,6 +700,7 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
 
         for (Long vpcId: vpcIds) {
             VpcVO vpc = _vpcDao.findById(vpcId);
+            // nothing to do if the VPC is not setup for distributed routing
             if (vpc == null || !vpc.usesDistributedRouter()) {
                 return;
             }
@@ -702,6 +710,9 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
             String bridgeName=generateBridgeNameForVpc(vpcId);
 
             OvsVpcPhysicalTopologyConfigCommand topologyConfigCommand = prepareVpcTopologyUpdate(vpcId);
+            topologyConfigCommand.setSequenceNumber(getNextSequenceNumber(vpcId));
+
+            // send topology change update to VPC spanned hosts
             for (Long id: vpcSpannedHostIds) {
                 if (!sendVpcTopologyChangeUpdate(topologyConfigCommand, id, bridgeName)) {
                     s_logger.debug("Failed to send VPC topology change update to host : " + id + ". Moving on " +
@@ -809,6 +820,10 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
                 if (network.getVpcId() != null && isVpcEnabledForDistributedRouter(network.getVpcId())) {
                     long vpcId = network.getVpcId();
                     OvsVpcRoutingPolicyConfigCommand cmd = prepareVpcRoutingPolicyUpdate(vpcId);
+                    cmd.setSequenceNumber(getNextSequenceNumber(vpcId));
+
+                    // get the list of hosts on which VPC spans (i.e hosts that need to be aware of VPC
+                    // network ACL update)
                     List<Long> vpcSpannedHostIds = _ovsNetworkToplogyGuru.getVpcSpannedHosts(vpcId);
                     for (Long id: vpcSpannedHostIds) {
                         if (!sendVpcRoutingPolicyChangeUpdate(cmd, id, bridgeName)) {
@@ -867,7 +882,7 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
         return cmd;
     }
 
-    public boolean sendVpcRoutingPolicyChangeUpdate(OvsVpcRoutingPolicyConfigCommand updateCmd, long hostId, String bridgeName) {
+    private boolean sendVpcRoutingPolicyChangeUpdate(OvsVpcRoutingPolicyConfigCommand updateCmd, long hostId, String bridgeName) {
         try {
             s_logger.debug("Sending VPC routing policies change update to the host " + hostId);
             updateCmd.setHostId(hostId);
@@ -885,4 +900,26 @@ public class OvsTunnelManagerImpl extends ManagerBase implements OvsTunnelManage
             return false;
         }
     }
+
+    private long getNextSequenceNumber(final long vpcId) {
+
+        try {
+            return  Transaction.execute(new TransactionCallback<Long>() {
+                @Override
+                public Long doInTransaction(TransactionStatus status) {
+                    VpcDistributedRouterSeqNoVO seqVo = _vpcDrSeqNoDao.findByVpcId(vpcId);
+                    if (seqVo == null) {
+                        seqVo = new VpcDistributedRouterSeqNoVO(vpcId);
+                        _vpcDrSeqNoDao.persist(seqVo);
+                    }
+                    _vpcDrSeqNoDao.lockRow(seqVo.getId(), true);
+                    seqVo.incrSequenceNo();
+                    _vpcDrSeqNoDao.update(seqVo.getId(), seqVo);
+                    return seqVo.getSequenceNo();
+                }
+            });
+        } finally {
+
+        }
+    }
 }

plugins/network-elements/ovs/src/com/cloud/network/ovs/dao/VpcDistributedRouterSeqNoDao.java

@@ -0,0 +1,23 @@
+// 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 com.cloud.network.ovs.dao;
+
+import com.cloud.utils.db.GenericDao;
+
+public interface VpcDistributedRouterSeqNoDao extends GenericDao<VpcDistributedRouterSeqNoVO, Long> {
+    VpcDistributedRouterSeqNoVO findByVpcId(long vpcId);
+}

plugins/network-elements/ovs/src/com/cloud/network/ovs/dao/VpcDistributedRouterSeqNoDaoImpl.java

@@ -0,0 +1,47 @@
+// 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 com.cloud.network.ovs.dao;
+
+import javax.ejb.Local;
+
+import org.apache.log4j.Logger;
+import org.springframework.stereotype.Component;
+
+import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+
+@Component
+@Local(value = {VpcDistributedRouterSeqNoDao.class})
+public class VpcDistributedRouterSeqNoDaoImpl extends GenericDaoBase<VpcDistributedRouterSeqNoVO, Long> implements VpcDistributedRouterSeqNoDao {
+    protected static final Logger s_logger = Logger.getLogger(VpcDistributedRouterSeqNoDaoImpl.class);
+    private SearchBuilder<VpcDistributedRouterSeqNoVO> VpcIdSearch;
+
+    protected VpcDistributedRouterSeqNoDaoImpl() {
+        VpcIdSearch = createSearchBuilder();
+        VpcIdSearch.and("vmId", VpcIdSearch.entity().getVpcId(), SearchCriteria.Op.EQ);
+        VpcIdSearch.done();
+    }
+
+    @Override
+    public VpcDistributedRouterSeqNoVO findByVpcId(long vpcId) {
+        SearchCriteria<VpcDistributedRouterSeqNoVO> sc = VpcIdSearch.create();
+        sc.setParameters("vmId", vpcId);
+        return findOneIncludingRemovedBy(sc);
+    }
+
+}

plugins/network-elements/ovs/src/com/cloud/network/ovs/dao/VpcDistributedRouterSeqNoVO.java

@@ -0,0 +1,72 @@
+// 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 com.cloud.network.ovs.dao;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+import org.apache.cloudstack.api.InternalIdentity;
+
+@Entity
+@Table(name = "op_vpc_distributed_router_sequence_no")
+public class VpcDistributedRouterSeqNoVO implements InternalIdentity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "id")
+    private Long id;
+
+    @Column(name = "vpc_id", updatable = false, nullable = false)
+    private Long vpcId;
+
+    @Column(name = "sequence_no")
+    long sequenceNo = 0;
+
+    protected VpcDistributedRouterSeqNoVO() {
+
+    }
+
+    public VpcDistributedRouterSeqNoVO(Long vpcId) {
+        super();
+        this.vpcId = vpcId;
+    }
+
+    @Override
+    public long getId() {
+        return id;
+    }
+
+    public Long getVpcId() {
+        return vpcId;
+    }
+
+    public void setVpcId(Long vpcId) {
+        this.vpcId = vpcId;
+    }
+
+    public long getSequenceNo() {
+        return sequenceNo;
+    }
+
+    public void incrSequenceNo() {
+        sequenceNo++;
+    }
+}

scripts/vm/hypervisor/xenserver/cloudstack_pluginlib.py

@@ -314,7 +314,7 @@ class jsonLoader(object):
         return '{%s}' % str(', '.join('%s : %s' % (k, repr(v)) for (k, v)
                                       in self.__dict__.iteritems()))
 
-def configure_bridge_for_network_topology(bridge, this_host_id, json_config):
+def configure_bridge_for_network_topology(bridge, this_host_id, json_config, sequence_no):
     vpconfig = jsonLoader(json.loads(json_config)).vpc
 
     if vpconfig is None:
@@ -389,7 +389,7 @@ def get_acl(vpcconfig, required_acl_id):
             return acl
     return None
 
-def configure_ovs_bridge_for_routing_policies(bridge, json_config):
+def configure_ovs_bridge_for_routing_policies(bridge, json_config, sequence_no):
     vpconfig = jsonLoader(json.loads(json_config)).vpc
 
     if vpconfig is None:

scripts/vm/hypervisor/xenserver/ovstunnel

@@ -400,15 +400,17 @@ def configure_ovs_bridge_for_network_topology(session, args):
     bridge = args.pop("bridge")
     json_config = args.pop("config")
     this_host_id = args.pop("host-id")
+    sequence_no = args.pop("seq-no")
 
-    return lib.configure_bridge_for_network_topology(bridge, this_host_id, json_config)
+    return lib.configure_bridge_for_network_topology(bridge, this_host_id, json_config, sequence_no)
 
 @echo
 def configure_ovs_bridge_for_routing_policies(session, args):
     bridge = args.pop("bridge")
     json_config = args.pop("config")
+    sequence_no = args.pop("seq-no")
 
-    return lib.configure_ovs_bridge_for_routing_policies(bridge, json_config)
+    return lib.configure_ovs_bridge_for_routing_policies(bridge, json_config, sequence_no)
 
 if __name__ == "__main__":
     XenAPIPlugin.dispatch({"create_tunnel": create_tunnel,

setup/db/db/schema-430to440.sql

@@ -770,6 +770,12 @@ ALTER TABLE `cloud`.`vpc_offerings` ADD COLUMN supports_region_level_vpc boolean
 ALTER TABLE `cloud`.`network_offerings` ADD COLUMN supports_streched_l2 boolean default false;
 ALTER TABLE `cloud`.`networks` ADD COLUMN streched_l2 boolean default false;
 ALTER TABLE `cloud`.`vpc` ADD COLUMN region_level_vpc boolean default false;
-
 INSERT INTO `cloud`.`configuration`(category, instance, component, name, value, description, default_value) VALUES ('Advanced', 'DEFAULT', 'NetworkOrchestrationService', 'router.redundant.vrrp.interval', '1', 'seconds between VRRP broadcast. It would 3 times broadcast fail to trigger fail-over mechanism of redundant router', '1') ON DUPLICATE KEY UPDATE category='Advanced';
 INSERT INTO `cloud`.`configuration`(category, instance, component, name, value, description, default_value) VALUES ('Advanced', 'DEFAULT', 'NetworkOrchestrationService', 'router.aggregation.command.each.timeout', '3', 'timeout in seconds for each Virtual Router command being aggregated. The final aggregation command timeout would be determined by this timeout * commands counts ', '3') ON DUPLICATE KEY UPDATE category='Advanced';
+CREATE TABLE `cloud`.`op_vpc_distributed_router_sequence_no` (
+  `id` bigint unsigned UNIQUE NOT NULL AUTO_INCREMENT COMMENT 'id',
+  `vpc_id` bigint unsigned NOT NULL COMMENT 'vpc id.',
+  `sequence_no` bigint unsigned  COMMENT 'seq number to be sent to agent, uniquely identifies topology or routing policy updates',
+  PRIMARY KEY (`id`),
+  UNIQUE `u_op_vpc_distributed_router_sequence_no_vpc_id`(`vpc_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8;