dmaap-bc/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 jakarta.ws.rs.core.Response.Status;
  26 import java.util.ArrayList;
  27 import java.util.Collection;
  28 import java.util.Iterator;
  29 import java.util.List;
  30 import java.util.Map;
  31 import java.util.Set;
  32 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
  33 import org.onap.dmaap.dbcapi.aaf.AafServiceFactory;
  34 import org.onap.dmaap.dbcapi.database.DatabaseClass;
  35 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
  36 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
  37 import org.onap.dmaap.dbcapi.model.ApiError;
  38 import org.onap.dmaap.dbcapi.model.DcaeLocation;
  39 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
  40 import org.onap.dmaap.dbcapi.model.MR_Client;
  41 import org.onap.dmaap.dbcapi.model.MR_Cluster;
  42 import org.onap.dmaap.dbcapi.model.MirrorMaker;
  43 import org.onap.dmaap.dbcapi.model.ReplicationType;
  44 import org.onap.dmaap.dbcapi.model.Topic;
  45 import org.onap.dmaap.dbcapi.util.DmaapConfig;
  46 import org.onap.dmaap.dbcapi.util.Fqdn;
  47 import org.onap.dmaap.dbcapi.util.Graph;
  48
  49 public class TopicService extends BaseLoggingClass {
  50
  51
  52     // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
  53     private static String defaultGlobalMrHost;
  54
  55     private Map<String, Topic> mr_topics;
  56
  57     private static DmaapService dmaapSvc = new DmaapService();
  58     private MR_ClientService clientService;
  59     private MR_ClusterService clusters;
  60     private DcaeLocationService locations;
  61     private MirrorMakerService bridge;
  62     private AafTopicSetupService aafTopicSetupService;
  63
  64     private static String centralCname;
  65     private boolean strictGraph = true;
  66     private boolean mmPerMR;
  67
  68
  69     public TopicService() {
  70         this(DatabaseClass.getTopics(), new MR_ClientService(), (DmaapConfig) DmaapConfig.getConfig(),
  71                 new MR_ClusterService(), new DcaeLocationService(), new MirrorMakerService(),
  72                 new AafTopicSetupService(
  73                         new AafServiceFactory().initAafService(ServiceType.AAF_TopicMgr),
  74                         dmaapSvc, (DmaapConfig) DmaapConfig.getConfig()));
  75     }
  76
  77     TopicService(Map<String, Topic> mr_topics, MR_ClientService clientService, DmaapConfig p,
  78                  MR_ClusterService clusters, DcaeLocationService locations, MirrorMakerService bridge, AafTopicSetupService aafTopicSetupService) {
  79         this.mr_topics = mr_topics;
  80         this.clientService = clientService;
  81         defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
  82         centralCname = p.getProperty("MR.CentralCname");
  83         String unit_test = p.getProperty("UnitTest", "No");
  84         if ("Yes".equals(unit_test)) {
  85             strictGraph = false;
  86         }
  87         mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
  88         logger.info("TopicService properties: CentralCname=" + centralCname +
  89                 "   defaultGlobarlMrHost=" + defaultGlobalMrHost +
  90                 " mmPerMR=" + mmPerMR);
  91         this.clusters = clusters;
  92         this.locations = locations;
  93         this.bridge = bridge;
  94         this.aafTopicSetupService = aafTopicSetupService;
  95     }
  96
  97     public Map<String, Topic> getTopics() {
  98         return mr_topics;
  99     }
 100
 101     public List<Topic> getAllTopics() {
 102         return getAllTopics(true);
 103     }
 104
 105     public List<Topic> getAllTopicsWithoutClients() {
 106         return getAllTopics(false);
 107     }
 108
 109     private List<Topic> getAllTopics(Boolean withClients) {
 110         ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
 111         if (withClients) {
 112             for (Topic topic : topics) {
 113                 topic.setClients(clientService.getAllMrClients(topic.getFqtn()));
 114             }
 115         }
 116         return topics;
 117     }
 118
 119     public Topic getTopic(String key, ApiError apiError) {
 120         logger.info("getTopic: key=" + key);
 121         Topic t = mr_topics.get(key);
 122         if (t == null) {
 123             apiError.setCode(Status.NOT_FOUND.getStatusCode());
 124             apiError.setFields("fqtn");
 125             apiError.setMessage("topic with fqtn " + key + " not found");
 126             return null;
 127         }
 128         t.setClients(clientService.getAllMrClients(key));
 129         apiError.setCode(Status.OK.getStatusCode());
 130         return t;
 131     }
 132
 133     public Topic addTopic(Topic topic, ApiError err, Boolean useExisting) {
 134         logger.info("Entry: addTopic");
 135         logger.info("Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle());
 136         String nFqtn = topic.genFqtn();
 137         logger.info("FQTN=" + nFqtn);
 138         Topic pTopic = getTopic(nFqtn, err);
 139         if (pTopic != null) {
 140             String t = "topic already exists: " + nFqtn;
 141             logger.info(t);
 142             if (useExisting) {
 143                 err.setCode(Status.OK.getStatusCode());
 144                 return pTopic;
 145             }
 146             err.setMessage(t);
 147             err.setFields("fqtn");
 148             err.setCode(Status.CONFLICT.getStatusCode());
 149             return null;
 150         }
 151         err.reset();  // err filled with NOT_FOUND is expected case, but don't want to litter...
 152
 153         topic.setFqtn(nFqtn);
 154
 155         ApiError topicSetupError = aafTopicSetupService.aafTopicSetup(topic);
 156         updateApiError(err, topicSetupError);
 157         if (err.getCode() >= 400) {
 158             return null;
 159         }
 160
 161         if (topic.getReplicationCase().involvesGlobal()) {
 162             if (topic.getGlobalMrURL() == null) {
 163                 topic.setGlobalMrURL(defaultGlobalMrHost);
 164             }
 165             if (!Fqdn.isValid(topic.getGlobalMrURL())) {
 166                 logger.error("GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
 167                 topic.setStatus(DmaapObject_Status.INVALID);
 168                 err.setCode(500);
 169                 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL());
 170                 err.setFields("globalMrURL");
 171
 172                 return null;
 173             }
 174         }
 175
 176
 177         if (topic.getNumClients() > 0) {
 178             ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
 179
 180
 181             ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
 182             for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
 183                 MR_Client c = it.next();
 184
 185                 logger.info("c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
 186                 MR_Client nc = new MR_Client(c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
 187                 nc.setFqtn(topic.getFqtn());
 188                 nc.setClientIdentity(c.getClientIdentity());
 189                 logger.info("nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
 190                 clients2.add(clientService.addMr_Client(nc, topic, err));
 191                 if (!err.is2xx()) {
 192                     return null;
 193                 }
 194             }
 195
 196             topic.setClients(clients2);
 197         }
 198
 199         Topic ntopic = checkForBridge(topic, err);
 200         if (ntopic == null) {
 201             topic.setStatus(DmaapObject_Status.INVALID);
 202             if (!err.is2xx()) {
 203                 return null;
 204             }
 205         }
 206
 207
 208         mr_topics.put(nFqtn, ntopic);
 209
 210         err.setCode(Status.OK.getStatusCode());
 211         return ntopic;
 212     }
 213
 214
 215     public Topic updateTopic(Topic topic, ApiError err) {
 216         logger.info("updateTopic: entry");
 217         logger.info("updateTopic: topic=" + topic);
 218         logger.info("updateTopic: fqtn=" + topic.getFqtn());
 219         if (topic.getFqtn().isEmpty()) {
 220             return null;
 221         }
 222         logger.info("updateTopic: call checkForBridge");
 223         Topic ntopic = checkForBridge(topic, err);
 224         if (ntopic == null) {
 225             topic.setStatus(DmaapObject_Status.INVALID);
 226             if (!err.is2xx()) {
 227                 return null;
 228             }
 229         }
 230         if (ntopic != null) {
 231             logger.info("updateTopic: call put");
 232             mr_topics.put(ntopic.getFqtn(), ntopic);
 233         }
 234         err.setCode(Status.OK.getStatusCode());
 235         return ntopic;
 236     }
 237
 238     public Topic removeTopic(String pubId, ApiError apiError) {
 239         Topic topic = mr_topics.get(pubId);
 240         if (topic == null) {
 241             apiError.setCode(Status.NOT_FOUND.getStatusCode());
 242             apiError.setMessage("Topic " + pubId + " does not exist");
 243             apiError.setFields("fqtn");
 244             return null;
 245         }
 246
 247         ApiError topicSetupError = aafTopicSetupService.aafTopicCleanup(topic);
 248         updateApiError(apiError, topicSetupError);
 249         if (apiError.getCode() >= 400) {
 250             return null;
 251         }
 252
 253         ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients(pubId));
 254         for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
 255             MR_Client c = it.next();
 256             clientService.removeMr_Client(c.getMrClientId(), false, apiError);
 257             if (!apiError.is2xx()) {
 258                 return null;
 259             }
 260         }
 261         apiError.setCode(Status.OK.getStatusCode());
 262         return mr_topics.remove(pubId);
 263     }
 264
 265     public static ApiError setBridgeClientPerms(MR_Cluster node) {
 266         DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
 267         String mmProvRole = p.getProperty("MM.ProvRole");
 268         String mmAgentRole = p.getProperty("MM.AgentRole");
 269         String[] Roles = {mmProvRole, mmAgentRole};
 270         String[] actions = {"view", "pub", "sub"};
 271         Topic bridgeAdminTopic = new Topic().init();
 272         bridgeAdminTopic.setTopicName(dmaapSvc.getBridgeAdminFqtn());
 273         bridgeAdminTopic.setTopicDescription("RESERVED topic for MirroMaker Provisioning");
 274         bridgeAdminTopic.setOwner("DBC");
 275
 276         ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
 277         for (String role : Roles) {
 278             MR_Client client = new MR_Client();
 279             client.setAction(actions);
 280             client.setClientRole(role);
 281             client.setDcaeLocationName(node.getDcaeLocationName());
 282             clients.add(client);
 283         }
 284         bridgeAdminTopic.setClients(clients);
 285
 286         TopicService ts = new TopicService();
 287         ApiError err = new ApiError();
 288         ts.addTopic(bridgeAdminTopic, err, true);
 289
 290         if (err.is2xx() || err.getCode() == 409) {
 291             err.setCode(200);
 292             return err;
 293         }
 294
 295         errorLogger.error(DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
 296         return err;
 297     }
 298
 299
 300     public Topic checkForBridge(Topic topic, ApiError err) {
 301         logger.info("checkForBridge: entry");
 302         logger.info("fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
 303         if (topic.getReplicationCase() == ReplicationType.REPLICATION_NONE) {
 304             topic.setStatus(DmaapObject_Status.VALID);
 305             return topic;
 306         }
 307
 308         boolean anythingWrong = false;
 309
 310         Set<String> groups = clusters.getGroups();
 311         for (String g : groups) {
 312             logger.info("buildBridge for " + topic.getFqtn() + " on group" + g);
 313             anythingWrong |= buildBridge(topic, err, g);
 314         }
 315         if (anythingWrong) {
 316             topic.setStatus(DmaapObject_Status.INVALID);
 317             if (!err.is2xx()) {
 318                 return null;
 319             }
 320         } else {
 321             topic.setStatus(DmaapObject_Status.VALID);
 322         }
 323         return topic;
 324     }
 325
 326     private boolean buildBridge(Topic topic, ApiError err, String group) {
 327         logger.info("buildBridge: entry");
 328         boolean anythingWrong = false;
 329         Graph graph;
 330         logger.info("buildBridge: strictGraph=" + strictGraph);
 331         if (group == null || group.isEmpty()) {
 332             graph = new Graph(topic.getClients(), strictGraph);
 333         } else {
 334             graph = new Graph(topic.getClients(), strictGraph, group);
 335         }
 336         logger.info("buildBridge: graph=" + graph);
 337         MR_Cluster groupCentralCluster = null;
 338
 339
 340         if (graph.isEmpty()) {
 341             logger.info("buildBridge: graph is empty.  return false");
 342             return false;
 343         } else if (group == null && topic.getReplicationCase().involvesFQDN()) {
 344             logger.info("buildBridge: group is null and replicationCaseInvolvesFQDN. return false");
 345             return false;
 346         } else if (!graph.hasCentral()) {
 347             logger.warn("Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
 348             return true;
 349         } else {
 350             groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
 351         }
 352         Collection<String> clientLocations = graph.getKeys();
 353         for (String loc : clientLocations) {
 354             logger.info("loc=" + loc);
 355             DcaeLocation location = locations.getDcaeLocation(loc);
 356             MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
 357             logger.info("cluster=" + cluster + " at " + cluster.getDcaeLocationName());
 358             logger.info("location.isCentral()=" + location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc());
 359
 360
 361             String source = null;
 362             String target = null;
 363
 364             /*
 365              * Provision Edge to Central bridges...
 366              */
 367             if (!location.isCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 368                 switch (topic.getReplicationCase()) {
 369                     case REPLICATION_EDGE_TO_CENTRAL:
 370                     case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:  // NOTE: this is for E2C portion only
 371                         source = cluster.getFqdn();
 372                         target = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
 373                         logger.info("REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" + target);
 374                         break;
 375                     case REPLICATION_CENTRAL_TO_EDGE:
 376                     case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for C2E portion only
 377                         source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
 378                         target = cluster.getFqdn();
 379                         break;
 380                     case REPLICATION_CENTRAL_TO_GLOBAL:
 381                     case REPLICATION_GLOBAL_TO_CENTRAL:
 382                     case REPLICATION_FQDN_TO_GLOBAL:
 383                     case REPLICATION_GLOBAL_TO_FQDN:
 384                         break;
 385
 386                     case REPLICATION_EDGE_TO_FQDN:
 387                     case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2C portion only
 388                         source = cluster.getFqdn();
 389                         target = groupCentralCluster.getFqdn();
 390                         break;
 391                     case REPLICATION_FQDN_TO_EDGE:
 392                     case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for F2E portion only
 393                         source = groupCentralCluster.getFqdn();
 394                         target = cluster.getFqdn();
 395                         break;
 396
 397                     default:
 398                         logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
 399                         anythingWrong = true;
 400                         err.setCode(400);
 401                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 402                                 + topic.getReplicationCase());
 403                         err.setMessage("Unexpected value for ReplicationType");
 404                         continue;
 405                 }
 406
 407             } else if (location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 408                 /*
 409                  * Provision Central to Global bridges
 410                  */
 411                 switch (topic.getReplicationCase()) {
 412
 413                     case REPLICATION_CENTRAL_TO_GLOBAL:
 414                     case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
 415                         source = centralCname;
 416                         target = topic.getGlobalMrURL();
 417                         break;
 418                     case REPLICATION_GLOBAL_TO_CENTRAL:
 419                     case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for G2C portion only
 420                         source = topic.getGlobalMrURL();
 421                         target = centralCname;
 422                         break;
 423
 424                     case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
 425                         source = groupCentralCluster.getFqdn();
 426                         target = topic.getGlobalMrURL();
 427                         break;
 428
 429                     case REPLICATION_FQDN_TO_GLOBAL:
 430                         source = groupCentralCluster.getFqdn();
 431                         target = topic.getGlobalMrURL();
 432                         break;
 433
 434                     case REPLICATION_GLOBAL_TO_FQDN:
 435                     case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for G2F portion only
 436                         source = topic.getGlobalMrURL();
 437                         target = groupCentralCluster.getFqdn();
 438                         break;
 439
 440                     case REPLICATION_FQDN_TO_EDGE:
 441                     case REPLICATION_EDGE_TO_FQDN:
 442                     case REPLICATION_EDGE_TO_CENTRAL:
 443                     case REPLICATION_CENTRAL_TO_EDGE:
 444                         break;
 445                     default:
 446                         logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
 447                         anythingWrong = true;
 448                         err.setCode(400);
 449                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 450                                 + topic.getReplicationCase());
 451                         err.setMessage("Unexpected value for ReplicationType");
 452                         continue;
 453                 }
 454             } else {
 455                 logger.warn("dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
 456                 anythingWrong = true;
 457                 continue;
 458             }
 459             if (source != null && target != null) {
 460                 try {
 461                     logger.info("Create a MM from " + source + " to " + target);
 462                     MirrorMaker mm = bridge.findNextMM(source, target, topic.getFqtn());
 463                     mm.addTopic(topic.getFqtn());
 464                     bridge.updateMirrorMaker(mm);
 465                 } catch (Exception ex) {
 466                     err.setCode(500);
 467                     err.setFields("mirror_maker.topic");
 468                     err.setMessage("Unexpected condition: " + ex);
 469                     anythingWrong = true;
 470                     break;
 471                 }
 472             }
 473
 474
 475         }
 476         return anythingWrong;
 477
 478     }
 479
 480
 481     /*
 482      * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
 483      * This was determined automatically based on presence of edge publishers and central subscribers.
 484      * The following method is a modification of that original logic, to preserve some backwards compatibility,
 485      * i.e. to be used when no ReplicationType is specified.
 486      */
 487
 488     public ReplicationType reviewTopic(Topic topic) {
 489
 490
 491         if (topic.getNumClients() > 1) {
 492             Graph graph = new Graph(topic.getClients(), false);
 493
 494             String centralFqdn = new String();
 495             if (graph.hasCentral()) {
 496                 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
 497                 centralFqdn = p.getProperty("MR.CentralCname");
 498             }
 499
 500             Collection<String> locations = graph.getKeys();
 501             for (String loc : locations) {
 502                 logger.info("loc=" + loc);
 503                 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
 504                 if (cluster == null) {
 505                     logger.info("No MR cluster for location " + loc);
 506                     continue;
 507                 }
 508                 if (graph.hasCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 509                     logger.info("Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn);
 510                     return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
 511
 512                 }
 513
 514             }
 515         }
 516
 517         return ReplicationType.REPLICATION_NONE;
 518     }
 519
 520     private void updateApiError(ApiError err, ApiError topicSetupError) {
 521         err.setCode(topicSetupError.getCode());
 522         err.setMessage(topicSetupError.getMessage());
 523         err.setFields(topicSetupError.getFields());
 524     }
 525 }