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.AafServiceImpl;
  27 import org.onap.dmaap.dbcapi.database.DatabaseClass;
  28 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
  29 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
  30 import org.onap.dmaap.dbcapi.model.ApiError;
  31 import org.onap.dmaap.dbcapi.model.DcaeLocation;
  32 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
  33 import org.onap.dmaap.dbcapi.model.MR_Client;
  34 import org.onap.dmaap.dbcapi.model.MR_Cluster;
  35 import org.onap.dmaap.dbcapi.model.MirrorMaker;
  36 import org.onap.dmaap.dbcapi.model.ReplicationType;
  37 import org.onap.dmaap.dbcapi.model.Topic;
  38 import org.onap.dmaap.dbcapi.util.DmaapConfig;
  39 import org.onap.dmaap.dbcapi.util.Fqdn;
  40 import org.onap.dmaap.dbcapi.util.Graph;
  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 AafServiceImpl(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         ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients(pubId));
 250         for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
 251             MR_Client c = it.next();
 252
 253
 254             clientService.removeMr_Client(c.getMrClientId(), false, apiError);
 255             if (!apiError.is2xx()) {
 256                 return null;
 257             }
 258         }
 259         apiError.setCode(Status.OK.getStatusCode());
 260         return mr_topics.remove(pubId);
 261     }
 262
 263     public static ApiError setBridgeClientPerms(MR_Cluster node) {
 264         DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
 265         String mmProvRole = p.getProperty("MM.ProvRole");
 266         String mmAgentRole = p.getProperty("MM.AgentRole");
 267         String[] Roles = {mmProvRole, mmAgentRole};
 268         String[] actions = {"view", "pub", "sub"};
 269         Topic bridgeAdminTopic = new Topic().init();
 270         bridgeAdminTopic.setTopicName(dmaapSvc.getBridgeAdminFqtn());
 271         bridgeAdminTopic.setTopicDescription("RESERVED topic for MirroMaker Provisioning");
 272         bridgeAdminTopic.setOwner("DBC");
 273
 274         ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
 275         for (String role : Roles) {
 276             MR_Client client = new MR_Client();
 277             client.setAction(actions);
 278             client.setClientRole(role);
 279             client.setDcaeLocationName(node.getDcaeLocationName());
 280             clients.add(client);
 281         }
 282         bridgeAdminTopic.setClients(clients);
 283
 284         TopicService ts = new TopicService();
 285         ApiError err = new ApiError();
 286         ts.addTopic(bridgeAdminTopic, err, true);
 287
 288         if (err.is2xx() || err.getCode() == 409) {
 289             err.setCode(200);
 290             return err;
 291         }
 292
 293         errorLogger.error(DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
 294         return err;
 295     }
 296
 297
 298     public Topic checkForBridge(Topic topic, ApiError err) {
 299         logger.info("checkForBridge: entry");
 300         logger.info("fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
 301         if (topic.getReplicationCase() == ReplicationType.REPLICATION_NONE) {
 302             topic.setStatus(DmaapObject_Status.VALID);
 303             return topic;
 304         }
 305
 306         boolean anythingWrong = false;
 307
 308         Set<String> groups = clusters.getGroups();
 309         for (String g : groups) {
 310             logger.info("buildBridge for " + topic.getFqtn() + " on group" + g);
 311             anythingWrong |= buildBridge(topic, err, g);
 312         }
 313         if (anythingWrong) {
 314             topic.setStatus(DmaapObject_Status.INVALID);
 315             if (!err.is2xx()) {
 316                 return null;
 317             }
 318         } else {
 319             topic.setStatus(DmaapObject_Status.VALID);
 320         }
 321         return topic;
 322     }
 323
 324     private boolean buildBridge(Topic topic, ApiError err, String group) {
 325         logger.info("buildBridge: entry");
 326         boolean anythingWrong = false;
 327         Graph graph;
 328         logger.info("buildBridge: strictGraph=" + strictGraph);
 329         if (group == null || group.isEmpty()) {
 330             graph = new Graph(topic.getClients(), strictGraph);
 331         } else {
 332             graph = new Graph(topic.getClients(), strictGraph, group);
 333         }
 334         logger.info("buildBridge: graph=" + graph);
 335         MR_Cluster groupCentralCluster = null;
 336
 337
 338         if (graph.isEmpty()) {
 339             logger.info("buildBridge: graph is empty.  return false");
 340             return false;
 341         } else if (group == null && topic.getReplicationCase().involvesFQDN()) {
 342             logger.info("buildBridge: group is null and replicationCaseInvolvesFQDN. return false");
 343             return false;
 344         } else if (!graph.hasCentral()) {
 345             logger.warn("Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
 346             return true;
 347         } else {
 348             groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
 349         }
 350         Collection<String> clientLocations = graph.getKeys();
 351         for (String loc : clientLocations) {
 352             logger.info("loc=" + loc);
 353             DcaeLocation location = locations.getDcaeLocation(loc);
 354             MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
 355             logger.info("cluster=" + cluster + " at " + cluster.getDcaeLocationName());
 356             logger.info("location.isCentral()=" + location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc());
 357
 358
 359             String source = null;
 360             String target = null;
 361
 362             /*
 363              * Provision Edge to Central bridges...
 364              */
 365             if (!location.isCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 366                 switch (topic.getReplicationCase()) {
 367                     case REPLICATION_EDGE_TO_CENTRAL:
 368                     case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:  // NOTE: this is for E2C portion only
 369                         source = cluster.getFqdn();
 370                         target = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
 371                         logger.info("REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" + target);
 372                         break;
 373                     case REPLICATION_CENTRAL_TO_EDGE:
 374                     case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for C2E portion only
 375                         source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
 376                         target = cluster.getFqdn();
 377                         break;
 378                     case REPLICATION_CENTRAL_TO_GLOBAL:
 379                     case REPLICATION_GLOBAL_TO_CENTRAL:
 380                     case REPLICATION_FQDN_TO_GLOBAL:
 381                     case REPLICATION_GLOBAL_TO_FQDN:
 382                         break;
 383
 384                     case REPLICATION_EDGE_TO_FQDN:
 385                     case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2C portion only
 386                         source = cluster.getFqdn();
 387                         target = groupCentralCluster.getFqdn();
 388                         break;
 389                     case REPLICATION_FQDN_TO_EDGE:
 390                     case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for F2E portion only
 391                         source = groupCentralCluster.getFqdn();
 392                         target = cluster.getFqdn();
 393                         break;
 394
 395                     default:
 396                         logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
 397                         anythingWrong = true;
 398                         err.setCode(400);
 399                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 400                                 + topic.getReplicationCase());
 401                         err.setMessage("Unexpected value for ReplicationType");
 402                         continue;
 403                 }
 404
 405             } else if (location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 406                 /*
 407                  * Provision Central to Global bridges
 408                  */
 409                 switch (topic.getReplicationCase()) {
 410
 411                     case REPLICATION_CENTRAL_TO_GLOBAL:
 412                     case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
 413                         source = centralCname;
 414                         target = topic.getGlobalMrURL();
 415                         break;
 416                     case REPLICATION_GLOBAL_TO_CENTRAL:
 417                     case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for G2C portion only
 418                         source = topic.getGlobalMrURL();
 419                         target = centralCname;
 420                         break;
 421
 422                     case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
 423                         source = groupCentralCluster.getFqdn();
 424                         target = topic.getGlobalMrURL();
 425                         break;
 426
 427                     case REPLICATION_FQDN_TO_GLOBAL:
 428                         source = groupCentralCluster.getFqdn();
 429                         target = topic.getGlobalMrURL();
 430                         break;
 431
 432                     case REPLICATION_GLOBAL_TO_FQDN:
 433                     case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for G2F portion only
 434                         source = topic.getGlobalMrURL();
 435                         target = groupCentralCluster.getFqdn();
 436                         break;
 437
 438                     case REPLICATION_FQDN_TO_EDGE:
 439                     case REPLICATION_EDGE_TO_FQDN:
 440                     case REPLICATION_EDGE_TO_CENTRAL:
 441                     case REPLICATION_CENTRAL_TO_EDGE:
 442                         break;
 443                     default:
 444                         logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
 445                         anythingWrong = true;
 446                         err.setCode(400);
 447                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 448                                 + topic.getReplicationCase());
 449                         err.setMessage("Unexpected value for ReplicationType");
 450                         continue;
 451                 }
 452             } else {
 453                 logger.warn("dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
 454                 anythingWrong = true;
 455                 continue;
 456             }
 457             if (source != null && target != null) {
 458                 try {
 459                     logger.info("Create a MM from " + source + " to " + target);
 460                     MirrorMaker mm = bridge.findNextMM(source, target, topic.getFqtn());
 461                     mm.addTopic(topic.getFqtn());
 462                     bridge.updateMirrorMaker(mm);
 463                 } catch (Exception ex) {
 464                     err.setCode(500);
 465                     err.setFields("mirror_maker.topic");
 466                     err.setMessage("Unexpected condition: " + ex);
 467                     anythingWrong = true;
 468                     break;
 469                 }
 470             }
 471
 472
 473         }
 474         return anythingWrong;
 475
 476     }
 477
 478
 479     /*
 480      * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
 481      * This was determined automatically based on presence of edge publishers and central subscribers.
 482      * The following method is a modification of that original logic, to preserve some backwards compatibility,
 483      * i.e. to be used when no ReplicationType is specified.
 484      */
 485
 486     public ReplicationType reviewTopic(Topic topic) {
 487
 488
 489         if (topic.getNumClients() > 1) {
 490             Graph graph = new Graph(topic.getClients(), false);
 491
 492             String centralFqdn = new String();
 493             if (graph.hasCentral()) {
 494                 DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
 495                 centralFqdn = p.getProperty("MR.CentralCname");
 496             }
 497
 498             Collection<String> locations = graph.getKeys();
 499             for (String loc : locations) {
 500                 logger.info("loc=" + loc);
 501                 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
 502                 if (cluster == null) {
 503                     logger.info("No MR cluster for location " + loc);
 504                     continue;
 505                 }
 506                 if (graph.hasCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 507                     logger.info("Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn);
 508                     return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
 509
 510                 }
 511
 512             }
 513         }
 514
 515         return ReplicationType.REPLICATION_NONE;
 516     }
 517
 518     private void updateApiError(ApiError err, ApiError topicSetupError) {
 519         err.setCode(topicSetupError.getCode());
 520         err.setMessage(topicSetupError.getMessage());
 521         err.setFields(topicSetupError.getFields());
 522     }
 523 }