1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
|
# Copyright 2019 CMCC Technologies Co., Ltd.
import json
import logging
import os
import time
import traceback
import uuid
from threading import Thread
from apscheduler.schedulers.background import BackgroundScheduler
from catalog.pub.database.models import NSPackageModel, VnfPackageModel
from catalog.pub.exceptions import CatalogException
from catalog.packages.biz import sdc_vnf_package
from catalog.pub.utils import fileutil
from catalog.packages.biz.ns_descriptor import NsDescriptor
from catalog.pub.utils.jobutil import JobUtil
from catalog.pub.utils.values import ignore_case_get
from catalog.pub.Dmaap_lib.dmaap.consumer import ConsumerClient
from catalog.pub.Dmaap_lib.dmaap.identity import IdentityClient
from catalog.pub.Dmaap_lib.dmaap.publisher import BatchPublisherClient
from catalog.pub.config.config import CONSUMER_GROUP, CONSUMER_ID, POLLING_INTERVAL, DMAAP_MR_BASE_URL
from catalog.pub.msapi import sdc
logger = logging.getLogger(__name__)
ARTIFACT_TYPES_LIST = ["TOSCA_TEMPLATE", "TOSCA_CSAR"]
class SDCController(Thread):
def __init__(self):
super(SDCController, self).__init__()
self.identity = IdentityClient(DMAAP_MR_BASE_URL)
self.scheduler = BackgroundScheduler(job_defaults={'max_instances': 3})
self.notification_topic = ''
self.status_topic = ''
self.consumer = ''
@self.scheduler.scheduled_job('interval', seconds=POLLING_INTERVAL)
def fetch_task():
self.fetch_notification()
def run(self):
try:
description = 'nfvo catalog key for' + CONSUMER_ID
key = self.identity.create_apikey('', description)
topics = sdc.register_for_topics(key['apiKey'])
self.notification_topic = topics['distrNotificationTopicName']
self.status_topic = topics['distrStatusTopicName']
self.consumer = ConsumerClient(DMAAP_MR_BASE_URL, self.notification_topic, CONSUMER_GROUP, CONSUMER_ID)
self.consumer.set_api_credentials(key['apiKey'], key['apiSecret'])
self.scheduler.start()
except Exception as e:
logger.error('start sdc controller failed.')
logger.error(str(e))
logger.error(traceback.format_exc())
def fetch_notification(self):
try:
logger.info('start to fetch message from dmaap.')
now_ms = int(time.time() * 1000)
notification_msgs = self.consumer.fetch()
logger.info('Receive a notification from dmaap: %s', notification_msgs)
for notification_msg in notification_msgs:
notification_callback = build_callback_notification(now_ms, notification_msg)
if is_activate_callback(notification_callback):
process_notification(notification_callback)
except Exception as e:
logger.error('fetch message from dmaap failed.')
logger.error(str(e))
logger.error(traceback.format_exc())
def is_activate_callback(notification_callback):
has_relevant_artifacts_in_resource = False
has_relevant_artifacts_in_service = False
if notification_callback['resources']:
has_relevant_artifacts_in_resource = True
if notification_callback['serviceArtifacts']:
has_relevant_artifacts_in_service = True
return has_relevant_artifacts_in_resource or has_relevant_artifacts_in_service
def build_callback_notification(now_ms, notification_msg):
# relevant_resource_instances = build_resource_instances(notification_msg, now_ms)
relevant_service_artifacts = handle_relevant_artifacts(notification_msg, now_ms,
notification_msg['serviceArtifacts'])
# notification_msg['resources'] = relevant_resource_instances
notification_msg['serviceArtifacts'] = relevant_service_artifacts
return notification_msg
def build_resource_instances(notification_msg, now_ms):
relevant_resource_instances = []
resources = notification_msg['resources']
for resource in resources:
artifacts = resource['artifacts']
found_relevant_artifacts = handle_relevant_artifacts(notification_msg, now_ms, artifacts)
if found_relevant_artifacts:
resources['artifacts'] = found_relevant_artifacts
relevant_resource_instances.append(resources)
return relevant_resource_instances
def handle_relevant_artifacts(notification_msg, now_ms, artifacts):
relevant_artifacts = []
for artifact in artifacts:
artifact_type = artifact['artifactType']
is_artifact_relevant = artifact_type in ARTIFACT_TYPES_LIST
if is_artifact_relevant:
generated_from_uuid = artifact.get('generatedFromUUID', '')
if generated_from_uuid:
generated_from_artifact = None
for artifact_g in artifacts:
if generated_from_uuid == artifact_g['artifactUUID']:
generated_from_artifact = artifact_g
break
if generated_from_artifact:
is_artifact_relevant = generated_from_artifact['artifactType'] in ARTIFACT_TYPES_LIST
else:
is_artifact_relevant = False
if is_artifact_relevant:
artifact = set_related_artifacts(artifact, notification_msg)
relevant_artifacts.append(artifact)
# notification_status = send_notification_status(now_ms, notification_msg['distributionID'], artifact, is_artifact_relevant)
# if notification_status != 'SUCCESS':
# logger.error("Error failed to send notification status to Dmaap.")
return relevant_artifacts
def set_related_artifacts(artifact, notification_msg):
related_artifacts_uuid = artifact.get('relatedArtifacts', '')
if related_artifacts_uuid:
related_artifacts = []
for artifact_uuid in related_artifacts_uuid:
related_artifacts.append(get_artifact_metadata(notification_msg, artifact_uuid))
artifact['relatedArtifactsInfo'] = related_artifacts
return artifact
def get_artifact_metadata(notification_msg, uuid):
service_artifacts = notification_msg['serviceArtifacts']
ret = None
for artifact in service_artifacts:
if artifact['artifactUUID'] == uuid:
ret = artifact
break
resources = notification_msg['resources']
if (not ret) and resources:
for resource in resources:
artifacts = resource['artifacts']
for artifact in artifacts:
if artifact['artifactUUID'] == uuid:
ret = artifact
break
if ret:
break
return ret
def send_notification_status(status_topic, now_ms, distribution_id, artifact, is_artifact_relevant):
logger.info('start to send notification status')
status = 'FAIL'
if is_artifact_relevant:
notification_status = 'NOTIFIED'
else:
notification_status = 'NOT_NOTIFIED'
request = {
'distributionID': distribution_id,
'consumerID': CONSUMER_ID,
'timestamp': now_ms,
'artifactURL': artifact['artifactURL'],
'status': notification_status
}
request_json = json.JSONEncoder().encode(request)
pub = BatchPublisherClient(DMAAP_MR_BASE_URL, status_topic, '', 'application/cambria')
logger.info('try to send notification status: %s', request_json)
try:
pub.send('MyPartitionKey', request_json)
time.sleep(1)
stuck = pub.close(10)
if not stuck:
status = 'SUCCESS'
logger.info('send notification status success.')
else:
logger.error('failed to send notification status, %s messages unsent', len(stuck))
except Exception as e:
logger.error('failed to send notification status.')
logger.error(str(e))
logger.error(traceback.format_exc())
return status
def process_notification(msg):
logger.info('Receive a callback notification, nb of resources: %s', len(msg['resources']))
try:
ns = sdc.get_asset(sdc.ASSETTYPE_SERVICES, msg['serviceUUID'])
# check if the related resources exist
resources = ns.get('resources', None)
job_array = []
resource_threads = []
if resources:
for resource in resources:
if (resource['resoucreType'] == 'VF') and not VnfPackageModel.objects.filter(vnfPackageId=resource['resourceUUID']):
logger.debug("VF [%s] is not distributed.", resource['resourceUUID'])
# raise CatalogException("VF (%s) is not distributed." % resource['resourceUUID'])
logger.info("VF [%s] begin to distribute.", resource['resourceUUID'])
csar_id = resource['resourceUUID']
vim_ids = ignore_case_get(resource, "vimIds")
lab_vim_id = ignore_case_get(resource, "labVimId")
job_id = str(uuid.uuid4())
job_array.append(job_id)
resource_threads.append(sdc_vnf_package.NfDistributeThread(csar_id, vim_ids, lab_vim_id, job_id))
# sdc_vnf_package.NfDistributeThread(csar_id, vim_ids, lab_vim_id, job_id).start()
else:
logger.debug("resource [%s] has been distributed", resource['resourceUUID'])
for resource_thread in resource_threads:
resource_thread.start()
for resource_thread in resource_threads:
resource_thread.join()
for jobID in job_array:
job_status = JobUtil.query_job_status(jobID)
if job_status[0].status == 'error':
raise CatalogException("VF resource fail to distributed.")
service_artifacts = msg['serviceArtifacts']
for artifact in service_artifacts:
if artifact['artifactType'] == 'TOSCA_CSAR':
csar_id = artifact['artifactUUID']
if not NSPackageModel.objects.filter(nsPackageId=artifact['artifactUUID']):
download_url = artifact['artifactURL']
localhost_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ns_csar_base = os.path.join(localhost_dir, "csars", "ns")
local_path = os.path.join(ns_csar_base, msg['distributionID'])
file_name = artifact['artifactName']
# csar_version = artifact['artifactVersion']
# download csar package
local_file_name = sdc.download_artifacts(download_url, local_path, file_name)
if local_file_name.endswith(".csar") or local_file_name.endswith(".zip"):
artifact_vnf_file = fileutil.unzip_file(local_file_name, local_path,
"Artifacts/Deployment/OTHER/ns.csar")
if os.path.exists(artifact_vnf_file):
local_file_name = artifact_vnf_file
data = {
'userDefinedData': {}
}
nsd = NsDescriptor()
nsd.create(data, csar_id)
nsd.parse_nsd_and_save(csar_id, local_file_name)
logger.debug("CSAR(%s) distriuted successfully.", csar_id)
else:
logger.debug("CSAR(%s) has been distriuted", csar_id)
except CatalogException as e:
logger.error("Failed to download the resource")
logger.error(str(e))
|
