src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * org.onap.dmaap
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  *
   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
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  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=========================================================
  21  */
  22
  23 package org.onap.dmaap.dbcapi.service;
  24
  25 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
  26 import org.onap.dmaap.dbcapi.aaf.AafServiceFactory;
  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;
  41
  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;
  47 import java.util.Map;
  48 import java.util.Set;
  49
  50 public class TopicService extends BaseLoggingClass {
  51
  52
  53     // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
  54     private static String defaultGlobalMrHost;
  55
  56     private Map<String, Topic> mr_topics;
  57
  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;
  64
  65     private static String centralCname;
  66     private boolean strictGraph = true;
  67     private boolean mmPerMR;
  68
  69
  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 AafServiceFactory().initAafService(ServiceType.AAF_TopicMgr),
  75                         dmaapSvc,
  76                         "true".equalsIgnoreCase(DmaapConfig.getConfig().getProperty("aaf.CreateTopicRoles", "true"))));
  77
  78     }
  79
  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)) {
  88             strictGraph = false;
  89         }
  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;
  96         this.bridge = bridge;
  97         this.aafTopicSetupService = aafTopicSetupService;
  98     }
  99
 100     public Map<String, Topic> getTopics() {
 101         return mr_topics;
 102     }
 103
 104     public List<Topic> getAllTopics() {
 105         return getAllTopics(true);
 106     }
 107
 108     public List<Topic> getAllTopicsWithoutClients() {
 109         return getAllTopics(false);
 110     }
 111
 112     private List<Topic> getAllTopics(Boolean withClients) {
 113         ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
 114         if (withClients) {
 115             for (Topic topic : topics) {
 116                 topic.setClients(clientService.getAllMrClients(topic.getFqtn()));
 117             }
 118         }
 119         return topics;
 120     }
 121
 122     public Topic getTopic(String key, ApiError apiError) {
 123         logger.info("getTopic: key=" + key);
 124         Topic t = mr_topics.get(key);
 125         if (t == null) {
 126             apiError.setCode(Status.NOT_FOUND.getStatusCode());
 127             apiError.setFields("fqtn");
 128             apiError.setMessage("topic with fqtn " + key + " not found");
 129             return null;
 130         }
 131         t.setClients(clientService.getAllMrClients(key));
 132         apiError.setCode(Status.OK.getStatusCode());
 133         return t;
 134     }
 135
 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;
 144             logger.info(t);
 145             if (useExisting) {
 146                 err.setCode(Status.OK.getStatusCode());
 147                 return pTopic;
 148             }
 149             err.setMessage(t);
 150             err.setFields("fqtn");
 151             err.setCode(Status.CONFLICT.getStatusCode());
 152             return null;
 153         }
 154         err.reset();  // err filled with NOT_FOUND is expected case, but don't want to litter...
 155
 156         topic.setFqtn(nFqtn);
 157
 158         ApiError topicSetupError = aafTopicSetupService.aafTopicSetup(topic);
 159         updateApiError(err, topicSetupError);
 160         if (err.getCode() >= 400) {
 161             return null;
 162         }
 163
 164         if (topic.getReplicationCase().involvesGlobal()) {
 165             if (topic.getGlobalMrURL() == null) {
 166                 topic.setGlobalMrURL(defaultGlobalMrHost);
 167             }
 168             if (!Fqdn.isValid(topic.getGlobalMrURL())) {
 169                 logger.error("GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
 170                 topic.setStatus(DmaapObject_Status.INVALID);
 171                 err.setCode(500);
 172                 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL());
 173                 err.setFields("globalMrURL");
 174
 175                 return null;
 176             }
 177         }
 178
 179
 180         if (topic.getNumClients() > 0) {
 181             ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
 182
 183
 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();
 187
 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));
 194                 if (!err.is2xx()) {
 195                     return null;
 196                 }
 197             }
 198
 199             topic.setClients(clients2);
 200         }
 201
 202         Topic ntopic = checkForBridge(topic, err);
 203         if (ntopic == null) {
 204             topic.setStatus(DmaapObject_Status.INVALID);
 205             if (!err.is2xx()) {
 206                 return null;
 207             }
 208         }
 209
 210
 211         mr_topics.put(nFqtn, ntopic);
 212
 213         err.setCode(Status.OK.getStatusCode());
 214         return ntopic;
 215     }
 216
 217
 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()) {
 223             return null;
 224         }
 225         logger.info("updateTopic: call checkForBridge");
 226         Topic ntopic = checkForBridge(topic, err);
 227         if (ntopic == null) {
 228             topic.setStatus(DmaapObject_Status.INVALID);
 229             if (!err.is2xx()) {
 230                 return null;
 231             }
 232         }
 233         if (ntopic != null) {
 234             logger.info("updateTopic: call put");
 235             mr_topics.put(ntopic.getFqtn(), ntopic);
 236         }
 237         err.setCode(Status.OK.getStatusCode());
 238         return ntopic;
 239     }
 240
 241     public Topic removeTopic(String pubId, ApiError apiError) {
 242         Topic topic = mr_topics.get(pubId);
 243         if (topic == null) {
 244             apiError.setCode(Status.NOT_FOUND.getStatusCode());
 245             apiError.setMessage("Topic " + pubId + " does not exist");
 246             apiError.setFields("fqtn");
 247             return null;
 248         }
 249
 250         ApiError topicSetupError = aafTopicSetupService.aafTopicCleanup(topic);
 251         updateApiError(apiError, topicSetupError);
 252         if (apiError.getCode() >= 400) {
 253             return null;
 254         }
 255
 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()) {
 261                 return null;
 262             }
 263         }
 264         apiError.setCode(Status.OK.getStatusCode());
 265         return mr_topics.remove(pubId);
 266     }
 267
 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");
 278
 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());
 285             clients.add(client);
 286         }
 287         bridgeAdminTopic.setClients(clients);
 288
 289         TopicService ts = new TopicService();
 290         ApiError err = new ApiError();
 291         ts.addTopic(bridgeAdminTopic, err, true);
 292
 293         if (err.is2xx() || err.getCode() == 409) {
 294             err.setCode(200);
 295             return err;
 296         }
 297
 298         errorLogger.error(DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
 299         return err;
 300     }
 301
 302
 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);
 308             return topic;
 309         }
 310
 311         boolean anythingWrong = false;
 312
 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);
 317         }
 318         if (anythingWrong) {
 319             topic.setStatus(DmaapObject_Status.INVALID);
 320             if (!err.is2xx()) {
 321                 return null;
 322             }
 323         } else {
 324             topic.setStatus(DmaapObject_Status.VALID);
 325         }
 326         return topic;
 327     }
 328
 329     private boolean buildBridge(Topic topic, ApiError err, String group) {
 330         logger.info("buildBridge: entry");
 331         boolean anythingWrong = false;
 332         Graph graph;
 333         logger.info("buildBridge: strictGraph=" + strictGraph);
 334         if (group == null || group.isEmpty()) {
 335             graph = new Graph(topic.getClients(), strictGraph);
 336         } else {
 337             graph = new Graph(topic.getClients(), strictGraph, group);
 338         }
 339         logger.info("buildBridge: graph=" + graph);
 340         MR_Cluster groupCentralCluster = null;
 341
 342
 343         if (graph.isEmpty()) {
 344             logger.info("buildBridge: graph is empty.  return false");
 345             return false;
 346         } else if (group == null && topic.getReplicationCase().involvesFQDN()) {
 347             logger.info("buildBridge: group is null and replicationCaseInvolvesFQDN. return false");
 348             return false;
 349         } else if (!graph.hasCentral()) {
 350             logger.warn("Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
 351             return true;
 352         } else {
 353             groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
 354         }
 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());
 362
 363
 364             String source = null;
 365             String target = null;
 366
 367             /*
 368              * Provision Edge to Central bridges...
 369              */
 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);
 377                         break;
 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();
 382                         break;
 383                     case REPLICATION_CENTRAL_TO_GLOBAL:
 384                     case REPLICATION_GLOBAL_TO_CENTRAL:
 385                     case REPLICATION_FQDN_TO_GLOBAL:
 386                     case REPLICATION_GLOBAL_TO_FQDN:
 387                         break;
 388
 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();
 393                         break;
 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();
 398                         break;
 399
 400                     default:
 401                         logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
 402                         anythingWrong = true;
 403                         err.setCode(400);
 404                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 405                                 + topic.getReplicationCase());
 406                         err.setMessage("Unexpected value for ReplicationType");
 407                         continue;
 408                 }
 409
 410             } else if (location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 411                 /*
 412                  * Provision Central to Global bridges
 413                  */
 414                 switch (topic.getReplicationCase()) {
 415
 416                     case REPLICATION_CENTRAL_TO_GLOBAL:
 417                     case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
 418                         source = centralCname;
 419                         target = topic.getGlobalMrURL();
 420                         break;
 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;
 425                         break;
 426
 427                     case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
 428                         source = groupCentralCluster.getFqdn();
 429                         target = topic.getGlobalMrURL();
 430                         break;
 431
 432                     case REPLICATION_FQDN_TO_GLOBAL:
 433                         source = groupCentralCluster.getFqdn();
 434                         target = topic.getGlobalMrURL();
 435                         break;
 436
 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();
 441                         break;
 442
 443                     case REPLICATION_FQDN_TO_EDGE:
 444                     case REPLICATION_EDGE_TO_FQDN:
 445                     case REPLICATION_EDGE_TO_CENTRAL:
 446                     case REPLICATION_CENTRAL_TO_EDGE:
 447                         break;
 448                     default:
 449                         logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
 450                         anythingWrong = true;
 451                         err.setCode(400);
 452                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 453                                 + topic.getReplicationCase());
 454                         err.setMessage("Unexpected value for ReplicationType");
 455                         continue;
 456                 }
 457             } else {
 458                 logger.warn("dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
 459                 anythingWrong = true;
 460                 continue;
 461             }
 462             if (source != null && target != null) {
 463                 try {
 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) {
 469                     err.setCode(500);
 470                     err.setFields("mirror_maker.topic");
 471                     err.setMessage("Unexpected condition: " + ex);
 472                     anythingWrong = true;
 473                     break;
 474                 }
 475             }
 476
 477
 478         }
 479         return anythingWrong;
 480
 481     }
 482
 483
 484     /*
 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.
 489      */
 490
 491     public ReplicationType reviewTopic(Topic topic) {
 492
 493
 494         if (topic.getNumClients() > 1) {
 495             Graph graph = new Graph(topic.getClients(), false);
 496
 497             String centralFqdn = new String();
 498             if (graph.hasCentral()) {
 499                 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
 500                 centralFqdn = p.getProperty("MR.CentralCname");
 501             }
 502
 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);
 509                     continue;
 510                 }
 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;
 514
 515                 }
 516
 517             }
 518         }
 519
 520         return ReplicationType.REPLICATION_NONE;
 521     }
 522
 523     private void updateApiError(ApiError err, ApiError topicSetupError) {
 524         err.setCode(topicSetupError.getCode());
 525         err.setMessage(topicSetupError.getMessage());
 526         err.setFields(topicSetupError.getFields());
 527     }
 528 }