get topology ids that need to be cleaned up
964020
private Set<String> get_cleanup_ids(StormClusterState clusterState, List<String> activeTopologies) throws Exception {
        List<String> task_ids = clusterState.task_storms();
        List<String> heartbeat_ids = clusterState.heartbeat_storms();
        List<String> error_ids = clusterState.task_error_storms();
        List<String> assignment_ids = clusterState.assignments(null);
        List<String> metric_ids = clusterState.get_metrics();

        HashSet<String> latest_code_ids = new HashSet<>();

        Set<String> code_ids = BlobStoreUtils.code_ids(nimbusData.getBlobStore());
        Set<String> to_cleanup_ids = new HashSet<>();
        Set<String> pendingTopologies = nimbusData.getPendingSubmitTopologies().buildMap().keySet();

        if (task_ids != null) {
            to_cleanup_ids.addAll(task_ids);
        }

        if (heartbeat_ids != null) {
            to_cleanup_ids.addAll(heartbeat_ids);
        }

        if (error_ids != null) {
            to_cleanup_ids.addAll(error_ids);
        }

        if (assignment_ids != null) {
            to_cleanup_ids.addAll(assignment_ids);
        }

        if (code_ids != null) {
            to_cleanup_ids.addAll(code_ids);
        }

        if (metric_ids != null) {
            to_cleanup_ids.addAll(metric_ids);
        }

        if (activeTopologies != null) {
            to_cleanup_ids.removeAll(activeTopologies);
            latest_code_ids.removeAll(activeTopologies);
        }

        to_cleanup_ids.removeAll(pendingTopologies);

        /**
         * Why need to remove latest code. Due to competition between Thrift.threads and TopologyAssign thread
         */
        to_cleanup_ids.removeAll(latest_code_ids);
        LOG.info("Skip removing topology of " + latest_code_ids);

        return to_cleanup_ids;
    }
740923
@Override
    public String getTopologyJsonPayload(Set<Host> activeHosts) {
		int count = NUM_RETRIER_ACROSS_NODES;
		String response;
		Exception lastEx = null;

		do {
			try {
				response = getTopologyFromRandomNodeWithRetry(activeHosts);
				if (response != null) {
					return response;
				}
			} catch (Exception e) {
				lastEx = e;
			} finally {
				count--;
			}
		} while ((count > 0));

		if (lastEx != null) {
			if (lastEx instanceof ConnectTimeoutException) {
				throw new TimeoutException("Unable to obtain topology", lastEx);
			}
			throw new DynoException(lastEx);
		} else {
			throw new DynoException("Could not contact dynomite for token map");
		}

    }
379924
public void cleanupDisappearedTopology() throws Exception {
        StormClusterState clusterState = nimbusData.getStormClusterState();

        List<String> activeTopologies = clusterState.active_storms();
        if (activeTopologies == null) {
            return;
        }

        Set<String> cleanupIds = get_cleanup_ids(clusterState, activeTopologies);
        for (String sysTopology : JStormMetrics.SYS_TOPOLOGIES) {
            cleanupIds.remove(sysTopology);
        }
        for (String topologyId : cleanupIds) {
            LOG.info("Cleaning up " + topologyId);
            clusterState.try_remove_storm(topologyId);
            nimbusData.getTaskHeartbeatsCache().remove(topologyId);
            nimbusData.getTasksHeartbeat().remove(topologyId);

            NimbusUtils.removeTopologyTaskTimeout(nimbusData, topologyId);

            // delete topology files in blobstore
            List<String> deleteKeys = BlobStoreUtils.getKeyListFromId(nimbusData, topologyId);
            BlobStoreUtils.cleanup_keys(deleteKeys, nimbusData.getBlobStore(), nimbusData.getStormClusterState());

            // don't need to delete local dir
        }
    }
678326
public static boolean verifyTopology(TopologyAPI.Topology topology) {
    if (!topology.hasName() || topology.getName().isEmpty()) {
      LOG.severe("Missing topology name");
      return false;
    }
    if (topology.getName().contains(".") || topology.getName().contains("/")) {
      LOG.severe("Invalid topology name. Topology name shouldn't have . or /");
      return false;
    }

    // Only verify RAM map string well-formed.
    getComponentRamMapConfig(topology);
    // Verify all bolts input streams exist. First get all output streams.
    Set<String> outputStreams = new HashSet<>();
    for (TopologyAPI.Spout spout : topology.getSpoutsList()) {
      for (TopologyAPI.OutputStream stream : spout.getOutputsList()) {
        outputStreams.add(
            stream.getStream().getComponentName() + "/" + stream.getStream().getId());
      }
    }
    for (TopologyAPI.Bolt bolt : topology.getBoltsList()) {
      for (TopologyAPI.OutputStream stream : bolt.getOutputsList()) {
        outputStreams.add(
            stream.getStream().getComponentName() + "/" + stream.getStream().getId());
      }
    }
    // Match output streams with input streams.
    for (TopologyAPI.Bolt bolt : topology.getBoltsList()) {
      for (TopologyAPI.InputStream stream : bolt.getInputsList()) {
        String key = stream.getStream().getComponentName() + "/" + stream.getStream().getId();
        if (!outputStreams.contains(key)) {
          LOG.severe("Invalid input stream " + key + " existing streams are " + outputStreams);
          return false;
        }
      }
    }
    // TODO(nbhagat): Should we enforce all output stream must be consumed?

    return true;
  }
577850
def clean_up_network(network_id, **kwargs):
    
    user_id = kwargs.get('user_id')
    #check_perm(user_id, 'delete_network')
    try:
        log.debug("Querying Network %s", network_id)
        net_i = db.DBSession.query(Network).filter(Network.id == network_id).\
        options(noload('scenarios')).options(noload('nodes')).options(noload('links')).options(noload('resourcegroups')).options(joinedload_all('types.templatetype.template')).one()
        net_i.attributes

        #Define the basic resource queries
        node_qry = db.DBSession.query(Node).filter(Node.network_id==network_id).filter(Node.status=='X').all()

        link_qry = db.DBSession.query(Link).filter(Link.network_id==network_id).filter(Link.status=='X').all()

        group_qry = db.DBSession.query(ResourceGroup).filter(ResourceGroup.network_id==network_id).filter(ResourceGroup.status=='X').all()

        scenario_qry = db.DBSession.query(Scenario).filter(Scenario.network_id==network_id).filter(Scenario.status=='X').all()


        for n in node_qry:
            db.DBSession.delete(n)
        for l in link_qry:
            db.DBSession.delete(l)
        for g in group_qry:
            db.DBSession.delete(g)
        for s in scenario_qry:
            db.DBSession.delete(s)

    except NoResultFound:
        raise ResourceNotFoundError("Network %s not found"%(network_id))
    db.DBSession.flush()
    return 'OK'
780287
private String[] getServerLayerIds(MapModel mapModel) {
		List<String> layerIds = new ArrayList<String>();
		for (Layer<?> layer : mapModel.getLayers()) {
			if (layer.isShowing()) {
				layerIds.add(layer.getServerLayerId());
			}
		}
		return layerIds.toArray(new String[layerIds.size()]);
	}
559442
def clean_probes(self, probe_ids):
        
        if isinstance(probe_ids, (tuple, list)):  # tuples & lists > x,y,z
            probe_ids = ",".join([str(_) for _ in probe_ids])

        return probe_ids
114550
def __IpsToServerIds(self):
    
    master_instance = self.__GetMasterInstance()
    assert(master_instance)
    retval, response = self.__RunMaprCli('node list -columns id')
    ip_to_id = {}
    for line_num, line in enumerate(response.split('\n')):
      tokens = line.split()
      if len(tokens) == 3 and tokens[0] != 'id':        
        instance_id = tokens[0]
        ip = tokens[2]
        ip_to_id[ip] = instance_id
    return ip_to_id
798240
function(transport) {
    return {
      /**
       * Get the server topology identifier.
       *
       * @returns {Number} Topology identifier.
       * @memberof Topology#
       * @since 0.3
       */
      getTopologyId: function() {
        return transport.getTopologyId();
      },
      /**
       * Get the list of servers that the client is currently connected to.
       *
       * @return {ServerAddress[]} An array of server addresses.
       * @memberof Topology#
       * @since 0.3
       */
      getMembers: function() {
        return transport.getMembers();
      },
      /**
       * Find the list of server addresses that are owners for a given key.
       *
       * @param {(String|Object)} k Key to find owners for.
       * @return {ServerAddress[]}
       * An array of server addresses that are owners for the given key.
       * @memberof Topology#
       * @since 0.3
       */
      findOwners: function(k) {
        return transport.findOwners(k);
      },
      /**
       * Switch remote cache manager to a different cluster,
       * previously declared via configuration.
       *
       * @param clusterName name of the cluster to which to switch to
       * @return {module:promise.Promise<Boolean>}
       * A promise encapsulating a Boolean that indicates {@code true} if the
       * switch happened, or {@code false} otherwise.
       * @memberof Topology#
       * @since 0.4
       */
      switchToCluster: function(clusterName) {
        return transport.switchToCluster(clusterName);
      },
      /**
       * Switch remote cache manager to the default cluster,
       * previously declared via configuration.
       *
       * @return {module:promise.Promise<Boolean>}
       * A promise encapsulating a Boolean that indicates {@code true} if the
       * switch happened, or {@code false} otherwise.
       * @memberof Topology#
       * @since 0.4
       */
      switchToDefaultCluster: function() {
        return transport.switchToDefaultCluster();
      }
    }
  }
791480
def _get_ids_from_hostname(self, hostname):
        """"""
        results = self.list_instances(hostname=hostname, mask="id")
        return [result['id'] for result in results]
593890
def _get_filtered_study_ids(shard, include_aliases=False):
    """"""
    from peyotl.phylesystem.helper import DIGIT_PATTERN
    k = shard.get_doc_ids()
    if shard.has_aliases and (not include_aliases):
        x = []
        for i in k:
            if DIGIT_PATTERN.match(i) or ((len(i) > 1) and (i[-2] == '_')):
                pass
            else:
                x.append(i)
        return x
1359614
def _get_ids_from_hostname(self, hostname):
        """"""
        results = self.list_hardware(hostname=hostname, mask="id")
        return [result['id'] for result in results]
109618
def get_auto_allocated_topology(self, project_id, **_params):
        """"""
        return self.get(
            self.auto_allocated_topology_path % project_id,
            params=_params)
406799
def _get_topology(self):
        
        self._topology.open()
        with self.__lock:
            self._kill_cursors_executor.open()
        return self._topology
653683
def _clean(self):
        """"""
        found_ids = {}
        nodes = [self._nodes[_node.Root.ID]]
        while nodes:
            node = nodes.pop()
            found_ids[node.id] = None
            nodes = nodes + node.children

        for node_id in self._nodes:
            if node_id in found_ids:
                continue
            logger.error('Dangling node: %s', node_id)

        for node_id in found_ids:
            if node_id in self._nodes:
                continue
            logger.error('Unregistered node: %s', node_id)
1043949
function () {
      var doUniqueConnectionIdsCheck = function (compound) {
        var duplicateConnectionIds = [];
        compound.connections.forEach(function (connection) {
          var connectionId = connection.connectionId;

          function filterByConnectionId (connection) {
            return connection.connectionId === connectionId;
          }

          var arrayByConnectionId = compound.connections.filter(filterByConnectionId);
          if (arrayByConnectionId.length > 1 && duplicateConnectionIds.indexOf(connectionId) < 0) {
            duplicateConnectionIds.push(connectionId);
          }
        });
        duplicateConnectionIds.forEach(function (connectionId) {
          errors.push('RuleViolation identified: Artifact \'' + compound.artifactId +
            '\' contains multiple connections with \'connectionId\'=\'' + connectionId + '\'');
        });
      };
      //
      if (doc.artifacts && doc.artifacts.compoundComponents) {
        doc.artifacts.compoundComponents.forEach(doUniqueConnectionIdsCheck);
      }
    }
211301
@VisibleForTesting
  void restartTopologyHandler(String topologyName) throws TopologyRuntimeManagementException {
    assert !potentialStaleExecutionData;
    Integer containerId = Context.topologyContainerId(config);
    Scheduler.RestartTopologyRequest restartTopologyRequest =
        Scheduler.RestartTopologyRequest.newBuilder()
            .setTopologyName(topologyName)
            .setContainerIndex(containerId)
            .build();
    // If we restart the container including TMaster, wee need to clean TMasterLocation,
    // since when starting up, TMaster expects no other existing TMaster,
    // i.e. TMasterLocation does not exist
    if (containerId == -1 || containerId == 0) {
      // get the instance of state manager to clean state
      SchedulerStateManagerAdaptor stateManager = Runtime.schedulerStateManagerAdaptor(runtime);
      Boolean result = stateManager.deleteTMasterLocation(topologyName);
      if (result == null || !result) {
        throw new TopologyRuntimeManagementException(
            "Failed to clear TMaster location. Check whether TMaster set it correctly.");
      }
    }

    if (!schedulerClient.restartTopology(restartTopologyRequest)) {
      throw new TopologyRuntimeManagementException(String.format(
          "Failed to restart topology '%s'", topologyName));
    }
    // Clean the connection when we are done.
    LOG.fine("Scheduler restarted topology successfully.");
  }