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