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
|
# Copyright 2019 CMCC Technologies Co., Ltd.
import json
import logging
import os
import time
import traceback
import uuid
from threading import Thread
from apscheduler.scheduler import Scheduler
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_IP, \
DMAAP_MR_PORT
from catalog.pub.msapi import sdc
logger = logging.getLogger(__name__)
DMAAP_MR_BASE_URL = "https://%s:%s" % (DMAAP_MR_IP, DMAAP_MR_PORT)
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 = Scheduler(standalone=True)
self.notification_topic = ''
self.status_topic = ''
self.consumer = ''
@self.scheduler.interval_schedule(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']))
service_artifacts = msg['serviceArtifacts']
for artifact in service_artifacts:
if artifact['artifactType'] == 'TOSCA_CSAR':
csar_id = 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']
sdc.download_artifacts(download_url, local_path, file_name)
# call ns package upload
data = {
'nsPackageId': csar_id,
'nsPackageVersion': csar_version,
'csarName': file_name,
'csarDir': local_path
}
jobid = uuid.uuid4()
# NsPackageParser(data, jobid).start()
logger.debug(data, jobid)
|
