-
Notifications
You must be signed in to change notification settings - Fork 23
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #831 from shtripat/expand-cluster-with-detected-peers
Added flow ExpandCluserWithDetectedPeers
- Loading branch information
Showing
2 changed files
with
156 additions
and
0 deletions.
There are no files selected for viewing
144 changes: 144 additions & 0 deletions
144
tendrl/commons/flows/expand_cluster_with_detected_peers/__init__.py
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 |
---|---|---|
@@ -0,0 +1,144 @@ | ||
import time | ||
import json | ||
import uuid | ||
|
||
import etcd | ||
|
||
from tendrl.commons import flows | ||
from tendrl.commons.flows.exceptions import FlowExecutionFailedError | ||
from tendrl.commons.objects.job import Job | ||
from tendrl.commons.utils import etcd_utils | ||
from tendrl.commons.utils import log_utils as logger | ||
|
||
|
||
class ExpandClusterWithDetectedPeers(flows.BaseFlow): | ||
def __init__(self, *args, **kwargs): | ||
super( | ||
ExpandClusterWithDetectedPeers, | ||
self | ||
).__init__(*args, **kwargs) | ||
|
||
def run(self): | ||
integration_id = self.parameters['TendrlContext.integration_id'] | ||
_cluster = NS.tendrl.objects.Cluster( | ||
integration_id=integration_id | ||
).load() | ||
if _cluster.status is not None and _cluster.status != "" and \ | ||
_cluster.status in ["importing", "unmanaging", "expanding"]: | ||
raise FlowExecutionFailedError( | ||
"Another job in progress for cluster, please wait till " | ||
"the job finishes (job_id: %s) (integration_id: %s) " % ( | ||
_cluster.current_job['job_id'], | ||
integration_id | ||
) | ||
) | ||
_lock_details = { | ||
'node_id': NS.node_context.node_id, | ||
'fqdn': NS.node_context.fqdn, | ||
'tags': NS.node_context.tags, | ||
'type': NS.type, | ||
'job_name': self.__class__.__name__, | ||
'job_id': self.job_id | ||
} | ||
_cluster.locked_by = _lock_details | ||
_cluster.status = "expanding" | ||
_cluster.current_job = { | ||
'job_id': self.job_id, | ||
'job_name': self.__class__.__name__, | ||
'status': 'in_progress' | ||
} | ||
_cluster.save() | ||
|
||
try: | ||
integration_id_index_key = \ | ||
"indexes/tags/tendrl/integration/%s" % integration_id | ||
node_ids = etcd_utils.read( | ||
integration_id_index_key).value | ||
node_ids = json.loads(node_ids) | ||
except etcd.EtcdKeyNotFound: | ||
raise FlowExecutionFailedError("Cluster with " | ||
"integration_id " | ||
"(%s) not found, cannot " | ||
"import" % integration_id) | ||
|
||
job_ids = [] | ||
for node_id in node_ids: | ||
_cnc = NS.tendrl.objects.ClusterNodeContext( | ||
node_id=node_id | ||
).load() | ||
if _cnc.is_managed.lower() == "yes": | ||
continue | ||
|
||
params = { | ||
'TendrlContext.integration_id': integration_id, | ||
'Node[]': [node_id], | ||
'Cluster.enable_volume_profiling': | ||
_cluster.enable_volume_profiling | ||
} | ||
payload = { | ||
"tags": ["tendrl/node_%s" % node_id], | ||
"run": "tendrl.flows.ImportCluster", | ||
"status": "new", | ||
"parent": self.parameters['job_id'], | ||
"parameters": params, | ||
"type": "node" | ||
} | ||
_job_id = str(uuid.uuid4()) | ||
Job(job_id=_job_id, status="new", payload=payload).save() | ||
logger.log( | ||
"info", | ||
NS.publisher_id, | ||
{ | ||
"message": "Importing (job: %s) Node %s " | ||
"to cluster %s" % ( | ||
_job_id, | ||
node_id, | ||
integration_id | ||
) | ||
}, | ||
job_id=self.parameters['job_id'] | ||
) | ||
job_ids.append(_job_id) | ||
|
||
loop_count = 0 | ||
# Wait for (no of nodes) * 6 minutes for import to complete | ||
wait_count = len(node_ids) * 36 | ||
while True: | ||
if loop_count >= wait_count: | ||
logger.log( | ||
"info", | ||
NS.publisher_id, | ||
{ | ||
"message": "Import jobs not yet complete " | ||
"on all new nodes. Timing out. (%s, %s)" % | ||
(str(node_ids), integration_id) | ||
}, | ||
job_id=self.parameters['job_id'], | ||
flow_id=self.parameters['flow_id'] | ||
) | ||
return False | ||
time.sleep(10) | ||
finished = True | ||
for job_id in job_ids: | ||
job = Job(job_id=job_id).load() | ||
if job.status != "finished": | ||
finished = False | ||
break | ||
if finished: | ||
break | ||
else: | ||
loop_count += 1 | ||
continue | ||
|
||
_cluster = NS.tendrl.objects.Cluster( | ||
integration_id=integration_id | ||
).load() | ||
_cluster.status = "" | ||
_cluster.locked_by = {} | ||
_cluster.current_job = { | ||
'status': "finished", | ||
'job_name': self.__class__.__name__, | ||
'job_id': self.job_id | ||
} | ||
_cluster.save() | ||
return True |
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