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.AafNamespace;
26 import org.onap.dmaap.dbcapi.aaf.AafRole;
27 import org.onap.dmaap.dbcapi.aaf.AafService;
28 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
29 import org.onap.dmaap.dbcapi.aaf.AafServiceImpl;
30 import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
31 import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
32 import org.onap.dmaap.dbcapi.database.DatabaseClass;
33 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
34 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
35 import org.onap.dmaap.dbcapi.model.ApiError;
36 import org.onap.dmaap.dbcapi.model.DcaeLocation;
37 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
38 import org.onap.dmaap.dbcapi.model.MR_Client;
39 import org.onap.dmaap.dbcapi.model.MR_Cluster;
40 import org.onap.dmaap.dbcapi.model.MirrorMaker;
41 import org.onap.dmaap.dbcapi.model.ReplicationType;
42 import org.onap.dmaap.dbcapi.model.Topic;
43 import org.onap.dmaap.dbcapi.util.DmaapConfig;
44 import org.onap.dmaap.dbcapi.util.Fqdn;
45 import org.onap.dmaap.dbcapi.util.Graph;
47 import javax.ws.rs.core.Response.Status;
48 import java.util.ArrayList;
49 import java.util.Collection;
50 import java.util.Iterator;
51 import java.util.List;
55 public class TopicService extends BaseLoggingClass {
58 // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
59 private static String defaultGlobalMrHost;
61 private Map<String, Topic> mr_topics;
63 private static DmaapService dmaapSvc = new DmaapService();
64 private MR_ClientService clientService;
65 private MR_ClusterService clusters;
66 private DcaeLocationService locations;
67 private MirrorMakerService bridge;
69 private static String centralCname;
70 private static boolean createTopicRoles;
71 private boolean strictGraph = true;
72 private boolean mmPerMR;
75 public TopicService() {
76 this(DatabaseClass.getTopics(), new MR_ClientService(), (DmaapConfig) DmaapConfig.getConfig(),
77 new MR_ClusterService(), new DcaeLocationService(), new MirrorMakerService());
81 TopicService(Map<String, Topic> mr_topics, MR_ClientService clientService, DmaapConfig p,
82 MR_ClusterService clusters, DcaeLocationService locations, MirrorMakerService bridge) {
83 this.mr_topics = mr_topics;
84 this.clientService = clientService;
85 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
86 centralCname = p.getProperty("MR.CentralCname");
87 createTopicRoles = "true".equalsIgnoreCase(p.getProperty("aaf.CreateTopicRoles", "true"));
88 String unit_test = p.getProperty("UnitTest", "No");
89 if ("Yes".equals(unit_test)) {
92 mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
93 logger.info("TopicService properties: CentralCname=" + centralCname +
94 " defaultGlobarlMrHost=" + defaultGlobalMrHost +
95 " createTopicRoles=" + createTopicRoles +
96 " mmPerMR=" + mmPerMR);
97 this.clusters = clusters;
98 this.locations = locations;
102 public Map<String, Topic> getTopics() {
106 public List<Topic> getAllTopics() {
107 return getAllTopics(true);
110 public List<Topic> getAllTopicsWithoutClients() {
111 return getAllTopics(false);
114 private List<Topic> getAllTopics(Boolean withClients) {
115 ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
117 for (Topic topic : topics) {
118 topic.setClients(clientService.getAllMrClients(topic.getFqtn()));
125 public Topic getTopic(String key, ApiError apiError) {
126 logger.info("getTopic: key=" + key);
127 Topic t = mr_topics.get(key);
129 apiError.setCode(Status.NOT_FOUND.getStatusCode());
130 apiError.setFields("fqtn");
131 apiError.setMessage("topic with fqtn " + key + " not found");
134 t.setClients(clientService.getAllMrClients(key));
135 apiError.setCode(Status.OK.getStatusCode());
139 private void aafTopicSetup(Topic topic, ApiError err) {
141 String nsr = dmaapSvc.getDmaap().getTopicNsRoot();
144 err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)");
145 err.setFields("topicNsRoot");
149 // establish AAF Connection using TopicMgr identity
150 AafService aaf = new AafServiceImpl(ServiceType.AAF_TopicMgr);
152 AafRole pubRole = null;
153 AafRole subRole = null;
155 // creating Topic Roles was not an original feature.
156 // For backwards compatibility, only do this if the feature is enabled.
157 // Also, if the namespace of the topic is a foreign namespace, (i.e. not the same as our root ns)
158 // then we likely don't have permission to create sub-ns and Roles so don't try.
159 if (createTopicRoles && topic.getFqtn().startsWith(nsr)) {
160 // create AAF namespace for this topic
161 AafNamespace ns = new AafNamespace(topic.getFqtn(), aaf.getIdentity());
163 int rc = aaf.addNamespace(ns);
164 if (rc != 201 && rc != 409) {
166 err.setMessage("Unexpected response from AAF:" + rc);
167 err.setFields("namespace:" + topic.getFqtn() + " identity=" + aaf.getIdentity());
172 // create AAF Roles for MR clients of this topic
173 String rn = "publisher";
174 pubRole = new AafRole(topic.getFqtn(), rn);
175 int rc = aaf.addRole(pubRole);
176 if (rc != 201 && rc != 409) {
178 err.setMessage("Unexpected response from AAF:" + rc);
179 err.setFields("topic:" + topic.getFqtn() + " role=" + rn);
182 topic.setPublisherRole(pubRole.getFullyQualifiedRole());
185 subRole = new AafRole(topic.getFqtn(), rn);
186 rc = aaf.addRole(subRole);
187 if (rc != 201 && rc != 409) {
189 err.setMessage("Unexpected response from AAF:" + rc);
190 err.setFields("topic:" + topic.getFqtn() + " role=" + rn);
193 topic.setSubscriberRole(subRole.getFullyQualifiedRole());
196 // create AAF perms checked by MR
197 String instance = ":topic." + topic.getFqtn();
198 String[] actions = {"pub", "sub", "view"};
199 String t = dmaapSvc.getTopicPerm();
200 for (String action : actions) {
201 DmaapPerm perm = new DmaapPerm(t, instance, action);
202 int rc = aaf.addPerm(perm);
203 if (rc != 201 && rc != 409) {
205 err.setMessage("Unexpected response from AAF:" + rc);
206 err.setFields("t=" + t + " instance=" + instance + " action=" + action);
209 if (createTopicRoles) {
210 // Grant perms to our default Roles
211 if (action.equals("pub") || action.equals("view")) {
212 DmaapGrant g = new DmaapGrant(perm, pubRole.getFullyQualifiedRole());
213 rc = aaf.addGrant(g);
214 if (rc != 201 && rc != 409) {
216 err.setMessage("Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole());
217 logger.warn(err.getMessage());
221 if (action.equals("sub") || action.equals("view")) {
222 DmaapGrant g = new DmaapGrant(perm, subRole.getFullyQualifiedRole());
223 rc = aaf.addGrant(g);
224 if (rc != 201 && rc != 409) {
226 err.setMessage("Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole());
227 logger.warn(err.getMessage());
236 public Topic addTopic(Topic topic, ApiError err, Boolean useExisting) {
237 logger.info("Entry: addTopic");
238 logger.info("Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle());
239 String nFqtn = topic.genFqtn();
240 logger.info("FQTN=" + nFqtn);
241 Topic pTopic = getTopic(nFqtn, err);
242 if (pTopic != null) {
243 String t = "topic already exists: " + nFqtn;
246 err.setCode(Status.OK.getStatusCode());
250 err.setFields("fqtn");
251 err.setCode(Status.CONFLICT.getStatusCode());
254 err.reset(); // err filled with NOT_FOUND is expected case, but don't want to litter...
256 topic.setFqtn(nFqtn);
258 aafTopicSetup(topic, err);
259 if (err.getCode() >= 400) {
263 if (topic.getReplicationCase().involvesGlobal()) {
264 if (topic.getGlobalMrURL() == null) {
265 topic.setGlobalMrURL(defaultGlobalMrHost);
267 if (!Fqdn.isValid(topic.getGlobalMrURL())) {
268 logger.error("GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
269 topic.setStatus(DmaapObject_Status.INVALID);
271 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL());
272 err.setFields("globalMrURL");
279 if (topic.getNumClients() > 0) {
280 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
283 ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
284 for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
285 MR_Client c = it.next();
287 logger.info("c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
288 MR_Client nc = new MR_Client(c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
289 nc.setFqtn(topic.getFqtn());
290 nc.setClientIdentity(c.getClientIdentity());
291 logger.info("nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
292 clients2.add(clientService.addMr_Client(nc, topic, err));
298 topic.setClients(clients2);
301 Topic ntopic = checkForBridge(topic, err);
302 if (ntopic == null) {
303 topic.setStatus(DmaapObject_Status.INVALID);
310 mr_topics.put(nFqtn, ntopic);
312 err.setCode(Status.OK.getStatusCode());
317 public Topic updateTopic(Topic topic, ApiError err) {
318 logger.info("updateTopic: entry");
319 logger.info("updateTopic: topic=" + topic);
320 logger.info("updateTopic: fqtn=" + topic.getFqtn());
321 if (topic.getFqtn().isEmpty()) {
324 logger.info("updateTopic: call checkForBridge");
325 Topic ntopic = checkForBridge(topic, err);
326 if (ntopic == null) {
327 topic.setStatus(DmaapObject_Status.INVALID);
332 if (ntopic != null) {
333 logger.info("updateTopic: call put");
334 mr_topics.put(ntopic.getFqtn(), ntopic);
336 err.setCode(Status.OK.getStatusCode());
340 public Topic removeTopic(String pubId, ApiError apiError) {
341 Topic topic = mr_topics.get(pubId);
343 apiError.setCode(Status.NOT_FOUND.getStatusCode());
344 apiError.setMessage("Topic " + pubId + " does not exist");
345 apiError.setFields("fqtn");
348 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients(pubId));
349 for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
350 MR_Client c = it.next();
353 clientService.removeMr_Client(c.getMrClientId(), false, apiError);
354 if (!apiError.is2xx()) {
358 apiError.setCode(Status.OK.getStatusCode());
359 return mr_topics.remove(pubId);
362 public static ApiError setBridgeClientPerms(MR_Cluster node) {
363 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
364 String mmProvRole = p.getProperty("MM.ProvRole");
365 String mmAgentRole = p.getProperty("MM.AgentRole");
366 String[] Roles = {mmProvRole, mmAgentRole};
367 String[] actions = {"view", "pub", "sub"};
368 Topic bridgeAdminTopic = new Topic().init();
369 bridgeAdminTopic.setTopicName(dmaapSvc.getBridgeAdminFqtn());
370 bridgeAdminTopic.setTopicDescription("RESERVED topic for MirroMaker Provisioning");
371 bridgeAdminTopic.setOwner("DBC");
373 ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
374 for (String role : Roles) {
375 MR_Client client = new MR_Client();
376 client.setAction(actions);
377 client.setClientRole(role);
378 client.setDcaeLocationName(node.getDcaeLocationName());
381 bridgeAdminTopic.setClients(clients);
383 TopicService ts = new TopicService();
384 ApiError err = new ApiError();
385 ts.addTopic(bridgeAdminTopic, err, true);
387 if (err.is2xx() || err.getCode() == 409) {
392 errorLogger.error(DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
397 public Topic checkForBridge(Topic topic, ApiError err) {
398 logger.info("checkForBridge: entry");
399 logger.info("fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
400 if (topic.getReplicationCase() == ReplicationType.REPLICATION_NONE) {
401 topic.setStatus(DmaapObject_Status.VALID);
405 boolean anythingWrong = false;
407 Set<String> groups = clusters.getGroups();
408 for (String g : groups) {
409 logger.info("buildBridge for " + topic.getFqtn() + " on group" + g);
410 anythingWrong |= buildBridge(topic, err, g);
413 topic.setStatus(DmaapObject_Status.INVALID);
418 topic.setStatus(DmaapObject_Status.VALID);
423 private boolean buildBridge(Topic topic, ApiError err, String group) {
424 logger.info("buildBridge: entry");
425 boolean anythingWrong = false;
427 logger.info("buildBridge: strictGraph=" + strictGraph);
428 if (group == null || group.isEmpty()) {
429 graph = new Graph(topic.getClients(), strictGraph);
431 graph = new Graph(topic.getClients(), strictGraph, group);
433 logger.info("buildBridge: graph=" + graph);
434 MR_Cluster groupCentralCluster = null;
437 if (graph.isEmpty()) {
438 logger.info("buildBridge: graph is empty. return false");
440 } else if (group == null && topic.getReplicationCase().involvesFQDN()) {
441 logger.info("buildBridge: group is null and replicationCaseInvolvesFQDN. return false");
443 } else if (!graph.hasCentral()) {
444 logger.warn("Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
447 groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
449 Collection<String> clientLocations = graph.getKeys();
450 for (String loc : clientLocations) {
451 logger.info("loc=" + loc);
452 DcaeLocation location = locations.getDcaeLocation(loc);
453 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
454 logger.info("cluster=" + cluster + " at " + cluster.getDcaeLocationName());
455 logger.info("location.isCentral()=" + location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc());
458 String source = null;
459 String target = null;
462 * Provision Edge to Central bridges...
464 if (!location.isCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
465 switch (topic.getReplicationCase()) {
466 case REPLICATION_EDGE_TO_CENTRAL:
467 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL: // NOTE: this is for E2C portion only
468 source = cluster.getFqdn();
469 target = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
470 logger.info("REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" + target);
472 case REPLICATION_CENTRAL_TO_EDGE:
473 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for C2E portion only
474 source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
475 target = cluster.getFqdn();
477 case REPLICATION_CENTRAL_TO_GLOBAL:
478 case REPLICATION_GLOBAL_TO_CENTRAL:
479 case REPLICATION_FQDN_TO_GLOBAL:
480 case REPLICATION_GLOBAL_TO_FQDN:
483 case REPLICATION_EDGE_TO_FQDN:
484 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2C portion only
485 source = cluster.getFqdn();
486 target = groupCentralCluster.getFqdn();
488 case REPLICATION_FQDN_TO_EDGE:
489 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for F2E portion only
490 source = groupCentralCluster.getFqdn();
491 target = cluster.getFqdn();
495 logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
496 anythingWrong = true;
498 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
499 + topic.getReplicationCase());
500 err.setMessage("Unexpected value for ReplicationType");
504 } else if (location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
506 * Provision Central to Global bridges
508 switch (topic.getReplicationCase()) {
510 case REPLICATION_CENTRAL_TO_GLOBAL:
511 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
512 source = centralCname;
513 target = topic.getGlobalMrURL();
515 case REPLICATION_GLOBAL_TO_CENTRAL:
516 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for G2C portion only
517 source = topic.getGlobalMrURL();
518 target = centralCname;
521 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2F portion only
522 source = groupCentralCluster.getFqdn();
523 target = topic.getGlobalMrURL();
526 case REPLICATION_FQDN_TO_GLOBAL:
527 source = groupCentralCluster.getFqdn();
528 target = topic.getGlobalMrURL();
531 case REPLICATION_GLOBAL_TO_FQDN:
532 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for G2F portion only
533 source = topic.getGlobalMrURL();
534 target = groupCentralCluster.getFqdn();
537 case REPLICATION_FQDN_TO_EDGE:
538 case REPLICATION_EDGE_TO_FQDN:
539 case REPLICATION_EDGE_TO_CENTRAL:
540 case REPLICATION_CENTRAL_TO_EDGE:
543 logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
544 anythingWrong = true;
546 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
547 + topic.getReplicationCase());
548 err.setMessage("Unexpected value for ReplicationType");
552 logger.warn("dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
553 anythingWrong = true;
556 if (source != null && target != null) {
558 logger.info("Create a MM from " + source + " to " + target);
559 MirrorMaker mm = bridge.findNextMM(source, target, topic.getFqtn());
560 mm.addTopic(topic.getFqtn());
561 bridge.updateMirrorMaker(mm);
562 } catch (Exception ex) {
564 err.setFields("mirror_maker.topic");
565 err.setMessage("Unexpected condition: " + ex);
566 anythingWrong = true;
573 return anythingWrong;
579 * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
580 * This was determined automatically based on presence of edge publishers and central subscribers.
581 * The following method is a modification of that original logic, to preserve some backwards compatibility,
582 * i.e. to be used when no ReplicationType is specified.
584 public ReplicationType reviewTopic(Topic topic) {
587 if (topic.getNumClients() > 1) {
588 Graph graph = new Graph(topic.getClients(), false);
590 String centralFqdn = new String();
591 if (graph.hasCentral()) {
592 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
593 centralFqdn = p.getProperty("MR.CentralCname");
596 Collection<String> locations = graph.getKeys();
597 for (String loc : locations) {
598 logger.info("loc=" + loc);
599 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
600 if (cluster == null) {
601 logger.info("No MR cluster for location " + loc);
604 if (graph.hasCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
605 logger.info("Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn);
606 return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
613 return ReplicationType.REPLICATION_NONE;