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