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