2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
7 * Modifications Copyright (C) 2019 IBM.
8 * ================================================================================
9 * Licensed under the Apache License, Version 2.0 (the "License");
10 * you may not use this file except in compliance with the License.
11 * You may obtain a copy of the License at
13 * http://www.apache.org/licenses/LICENSE-2.0
15 * Unless required by applicable law or agreed to in writing, software
16 * distributed under the License is distributed on an "AS IS" BASIS,
17 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18 * See the License for the specific language governing permissions and
19 * limitations under the License.
20 * ============LICENSE_END=========================================================
23 package org.onap.dmaap.dbcapi.service;
25 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
26 import org.onap.dmaap.dbcapi.aaf.AafServiceImpl;
27 import org.onap.dmaap.dbcapi.database.DatabaseClass;
28 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
29 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
30 import org.onap.dmaap.dbcapi.model.ApiError;
31 import org.onap.dmaap.dbcapi.model.DcaeLocation;
32 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
33 import org.onap.dmaap.dbcapi.model.MR_Client;
34 import org.onap.dmaap.dbcapi.model.MR_Cluster;
35 import org.onap.dmaap.dbcapi.model.MirrorMaker;
36 import org.onap.dmaap.dbcapi.model.ReplicationType;
37 import org.onap.dmaap.dbcapi.model.Topic;
38 import org.onap.dmaap.dbcapi.util.DmaapConfig;
39 import org.onap.dmaap.dbcapi.util.Fqdn;
40 import org.onap.dmaap.dbcapi.util.Graph;
42 import javax.ws.rs.core.Response.Status;
43 import java.util.ArrayList;
44 import java.util.Collection;
45 import java.util.Iterator;
46 import java.util.List;
50 public class TopicService extends BaseLoggingClass {
53 // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
54 private static String defaultGlobalMrHost;
56 private Map<String, Topic> mr_topics;
58 private static DmaapService dmaapSvc = new DmaapService();
59 private MR_ClientService clientService;
60 private MR_ClusterService clusters;
61 private DcaeLocationService locations;
62 private MirrorMakerService bridge;
63 private AafTopicSetupService aafTopicSetupService;
65 private static String centralCname;
66 private boolean strictGraph = true;
67 private boolean mmPerMR;
70 public TopicService() {
71 this(DatabaseClass.getTopics(), new MR_ClientService(), (DmaapConfig) DmaapConfig.getConfig(),
72 new MR_ClusterService(), new DcaeLocationService(), new MirrorMakerService(),
73 new AafTopicSetupService(
74 new AafServiceImpl(ServiceType.AAF_TopicMgr),
76 "true".equalsIgnoreCase(DmaapConfig.getConfig().getProperty("aaf.CreateTopicRoles", "true"))));
80 TopicService(Map<String, Topic> mr_topics, MR_ClientService clientService, DmaapConfig p,
81 MR_ClusterService clusters, DcaeLocationService locations, MirrorMakerService bridge, AafTopicSetupService aafTopicSetupService) {
82 this.mr_topics = mr_topics;
83 this.clientService = clientService;
84 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
85 centralCname = p.getProperty("MR.CentralCname");
86 String unit_test = p.getProperty("UnitTest", "No");
87 if ("Yes".equals(unit_test)) {
90 mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
91 logger.info("TopicService properties: CentralCname=" + centralCname +
92 " defaultGlobarlMrHost=" + defaultGlobalMrHost +
93 " mmPerMR=" + mmPerMR);
94 this.clusters = clusters;
95 this.locations = locations;
97 this.aafTopicSetupService = aafTopicSetupService;
100 public Map<String, Topic> getTopics() {
104 public List<Topic> getAllTopics() {
105 return getAllTopics(true);
108 public List<Topic> getAllTopicsWithoutClients() {
109 return getAllTopics(false);
112 private List<Topic> getAllTopics(Boolean withClients) {
113 ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
115 for (Topic topic : topics) {
116 topic.setClients(clientService.getAllMrClients(topic.getFqtn()));
122 public Topic getTopic(String key, ApiError apiError) {
123 logger.info("getTopic: key=" + key);
124 Topic t = mr_topics.get(key);
126 apiError.setCode(Status.NOT_FOUND.getStatusCode());
127 apiError.setFields("fqtn");
128 apiError.setMessage("topic with fqtn " + key + " not found");
131 t.setClients(clientService.getAllMrClients(key));
132 apiError.setCode(Status.OK.getStatusCode());
136 public Topic addTopic(Topic topic, ApiError err, Boolean useExisting) {
137 logger.info("Entry: addTopic");
138 logger.info("Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle());
139 String nFqtn = topic.genFqtn();
140 logger.info("FQTN=" + nFqtn);
141 Topic pTopic = getTopic(nFqtn, err);
142 if (pTopic != null) {
143 String t = "topic already exists: " + nFqtn;
146 err.setCode(Status.OK.getStatusCode());
150 err.setFields("fqtn");
151 err.setCode(Status.CONFLICT.getStatusCode());
154 err.reset(); // err filled with NOT_FOUND is expected case, but don't want to litter...
156 topic.setFqtn(nFqtn);
158 ApiError topicSetupError = aafTopicSetupService.aafTopicSetup(topic);
159 updateApiError(err, topicSetupError);
160 if (err.getCode() >= 400) {
164 if (topic.getReplicationCase().involvesGlobal()) {
165 if (topic.getGlobalMrURL() == null) {
166 topic.setGlobalMrURL(defaultGlobalMrHost);
168 if (!Fqdn.isValid(topic.getGlobalMrURL())) {
169 logger.error("GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
170 topic.setStatus(DmaapObject_Status.INVALID);
172 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL());
173 err.setFields("globalMrURL");
180 if (topic.getNumClients() > 0) {
181 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
184 ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
185 for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
186 MR_Client c = it.next();
188 logger.info("c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
189 MR_Client nc = new MR_Client(c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
190 nc.setFqtn(topic.getFqtn());
191 nc.setClientIdentity(c.getClientIdentity());
192 logger.info("nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
193 clients2.add(clientService.addMr_Client(nc, topic, err));
199 topic.setClients(clients2);
202 Topic ntopic = checkForBridge(topic, err);
203 if (ntopic == null) {
204 topic.setStatus(DmaapObject_Status.INVALID);
211 mr_topics.put(nFqtn, ntopic);
213 err.setCode(Status.OK.getStatusCode());
218 public Topic updateTopic(Topic topic, ApiError err) {
219 logger.info("updateTopic: entry");
220 logger.info("updateTopic: topic=" + topic);
221 logger.info("updateTopic: fqtn=" + topic.getFqtn());
222 if (topic.getFqtn().isEmpty()) {
225 logger.info("updateTopic: call checkForBridge");
226 Topic ntopic = checkForBridge(topic, err);
227 if (ntopic == null) {
228 topic.setStatus(DmaapObject_Status.INVALID);
233 if (ntopic != null) {
234 logger.info("updateTopic: call put");
235 mr_topics.put(ntopic.getFqtn(), ntopic);
237 err.setCode(Status.OK.getStatusCode());
241 public Topic removeTopic(String pubId, ApiError apiError) {
242 Topic topic = mr_topics.get(pubId);
244 apiError.setCode(Status.NOT_FOUND.getStatusCode());
245 apiError.setMessage("Topic " + pubId + " does not exist");
246 apiError.setFields("fqtn");
250 ApiError topicSetupError = aafTopicSetupService.aafTopicCleanup(topic);
251 updateApiError(apiError, topicSetupError);
252 if (apiError.getCode() >= 400) {
256 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients(pubId));
257 for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
258 MR_Client c = it.next();
259 clientService.removeMr_Client(c.getMrClientId(), false, apiError);
260 if (!apiError.is2xx()) {
264 apiError.setCode(Status.OK.getStatusCode());
265 return mr_topics.remove(pubId);
268 public static ApiError setBridgeClientPerms(MR_Cluster node) {
269 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
270 String mmProvRole = p.getProperty("MM.ProvRole");
271 String mmAgentRole = p.getProperty("MM.AgentRole");
272 String[] Roles = {mmProvRole, mmAgentRole};
273 String[] actions = {"view", "pub", "sub"};
274 Topic bridgeAdminTopic = new Topic().init();
275 bridgeAdminTopic.setTopicName(dmaapSvc.getBridgeAdminFqtn());
276 bridgeAdminTopic.setTopicDescription("RESERVED topic for MirroMaker Provisioning");
277 bridgeAdminTopic.setOwner("DBC");
279 ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
280 for (String role : Roles) {
281 MR_Client client = new MR_Client();
282 client.setAction(actions);
283 client.setClientRole(role);
284 client.setDcaeLocationName(node.getDcaeLocationName());
287 bridgeAdminTopic.setClients(clients);
289 TopicService ts = new TopicService();
290 ApiError err = new ApiError();
291 ts.addTopic(bridgeAdminTopic, err, true);
293 if (err.is2xx() || err.getCode() == 409) {
298 errorLogger.error(DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
303 public Topic checkForBridge(Topic topic, ApiError err) {
304 logger.info("checkForBridge: entry");
305 logger.info("fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
306 if (topic.getReplicationCase() == ReplicationType.REPLICATION_NONE) {
307 topic.setStatus(DmaapObject_Status.VALID);
311 boolean anythingWrong = false;
313 Set<String> groups = clusters.getGroups();
314 for (String g : groups) {
315 logger.info("buildBridge for " + topic.getFqtn() + " on group" + g);
316 anythingWrong |= buildBridge(topic, err, g);
319 topic.setStatus(DmaapObject_Status.INVALID);
324 topic.setStatus(DmaapObject_Status.VALID);
329 private boolean buildBridge(Topic topic, ApiError err, String group) {
330 logger.info("buildBridge: entry");
331 boolean anythingWrong = false;
333 logger.info("buildBridge: strictGraph=" + strictGraph);
334 if (group == null || group.isEmpty()) {
335 graph = new Graph(topic.getClients(), strictGraph);
337 graph = new Graph(topic.getClients(), strictGraph, group);
339 logger.info("buildBridge: graph=" + graph);
340 MR_Cluster groupCentralCluster = null;
343 if (graph.isEmpty()) {
344 logger.info("buildBridge: graph is empty. return false");
346 } else if (group == null && topic.getReplicationCase().involvesFQDN()) {
347 logger.info("buildBridge: group is null and replicationCaseInvolvesFQDN. return false");
349 } else if (!graph.hasCentral()) {
350 logger.warn("Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
353 groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
355 Collection<String> clientLocations = graph.getKeys();
356 for (String loc : clientLocations) {
357 logger.info("loc=" + loc);
358 DcaeLocation location = locations.getDcaeLocation(loc);
359 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
360 logger.info("cluster=" + cluster + " at " + cluster.getDcaeLocationName());
361 logger.info("location.isCentral()=" + location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc());
364 String source = null;
365 String target = null;
368 * Provision Edge to Central bridges...
370 if (!location.isCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
371 switch (topic.getReplicationCase()) {
372 case REPLICATION_EDGE_TO_CENTRAL:
373 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL: // NOTE: this is for E2C portion only
374 source = cluster.getFqdn();
375 target = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
376 logger.info("REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" + target);
378 case REPLICATION_CENTRAL_TO_EDGE:
379 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for C2E portion only
380 source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
381 target = cluster.getFqdn();
383 case REPLICATION_CENTRAL_TO_GLOBAL:
384 case REPLICATION_GLOBAL_TO_CENTRAL:
385 case REPLICATION_FQDN_TO_GLOBAL:
386 case REPLICATION_GLOBAL_TO_FQDN:
389 case REPLICATION_EDGE_TO_FQDN:
390 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2C portion only
391 source = cluster.getFqdn();
392 target = groupCentralCluster.getFqdn();
394 case REPLICATION_FQDN_TO_EDGE:
395 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for F2E portion only
396 source = groupCentralCluster.getFqdn();
397 target = cluster.getFqdn();
401 logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
402 anythingWrong = true;
404 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
405 + topic.getReplicationCase());
406 err.setMessage("Unexpected value for ReplicationType");
410 } else if (location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
412 * Provision Central to Global bridges
414 switch (topic.getReplicationCase()) {
416 case REPLICATION_CENTRAL_TO_GLOBAL:
417 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
418 source = centralCname;
419 target = topic.getGlobalMrURL();
421 case REPLICATION_GLOBAL_TO_CENTRAL:
422 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for G2C portion only
423 source = topic.getGlobalMrURL();
424 target = centralCname;
427 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2F portion only
428 source = groupCentralCluster.getFqdn();
429 target = topic.getGlobalMrURL();
432 case REPLICATION_FQDN_TO_GLOBAL:
433 source = groupCentralCluster.getFqdn();
434 target = topic.getGlobalMrURL();
437 case REPLICATION_GLOBAL_TO_FQDN:
438 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for G2F portion only
439 source = topic.getGlobalMrURL();
440 target = groupCentralCluster.getFqdn();
443 case REPLICATION_FQDN_TO_EDGE:
444 case REPLICATION_EDGE_TO_FQDN:
445 case REPLICATION_EDGE_TO_CENTRAL:
446 case REPLICATION_CENTRAL_TO_EDGE:
449 logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
450 anythingWrong = true;
452 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
453 + topic.getReplicationCase());
454 err.setMessage("Unexpected value for ReplicationType");
458 logger.warn("dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
459 anythingWrong = true;
462 if (source != null && target != null) {
464 logger.info("Create a MM from " + source + " to " + target);
465 MirrorMaker mm = bridge.findNextMM(source, target, topic.getFqtn());
466 mm.addTopic(topic.getFqtn());
467 bridge.updateMirrorMaker(mm);
468 } catch (Exception ex) {
470 err.setFields("mirror_maker.topic");
471 err.setMessage("Unexpected condition: " + ex);
472 anythingWrong = true;
479 return anythingWrong;
485 * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
486 * This was determined automatically based on presence of edge publishers and central subscribers.
487 * The following method is a modification of that original logic, to preserve some backwards compatibility,
488 * i.e. to be used when no ReplicationType is specified.
491 public ReplicationType reviewTopic(Topic topic) {
494 if (topic.getNumClients() > 1) {
495 Graph graph = new Graph(topic.getClients(), false);
497 String centralFqdn = new String();
498 if (graph.hasCentral()) {
499 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
500 centralFqdn = p.getProperty("MR.CentralCname");
503 Collection<String> locations = graph.getKeys();
504 for (String loc : locations) {
505 logger.info("loc=" + loc);
506 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
507 if (cluster == null) {
508 logger.info("No MR cluster for location " + loc);
511 if (graph.hasCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
512 logger.info("Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn);
513 return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
520 return ReplicationType.REPLICATION_NONE;
523 private void updateApiError(ApiError err, ApiError topicSetupError) {
524 err.setCode(topicSetupError.getCode());
525 err.setMessage(topicSetupError.getMessage());
526 err.setFields(topicSetupError.getFields());