src/main/java/org/onap/dmaap/dbcapi/service/MirrorMakerService.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.List;
  25 import java.util.Map;
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35 //import org.openecomp.dmaapbc.aaf.AndrewDecryptor;
  36 import org.onap.dmaap.dbcapi.aaf.AafDecrypt;
  37 import org.onap.dmaap.dbcapi.client.MrTopicConnection;
  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.MR_Cluster;
  43 import org.onap.dmaap.dbcapi.model.MirrorMaker;
  44 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
  45 import org.onap.dmaap.dbcapi.util.DmaapConfig;
  46 import org.onap.dmaap.dbcapi.util.RandomInteger;
  47
  48 public class MirrorMakerService extends BaseLoggingClass {
  49
  50         private Map<String, MirrorMaker> mirrors = DatabaseClass.getMirrorMakers();
  51         private static MrTopicConnection prov;
  52         private static AafDecrypt decryptor;
  53
  54         private static String provUser;
  55         private static String provUserPwd;
  56         private static String defaultProducerPort;
  57         private static String defaultConsumerPort;
  58         private static String centralFqdn;
  59         private int maxTopicsPerMM;
  60         private boolean mmPerMR;
  61
  62         public MirrorMakerService() {
  63                 super();
  64                 decryptor = new AafDecrypt();
  65                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
  66                 provUser = p.getProperty("MM.ProvUserMechId");
  67                 provUserPwd = decryptor.decrypt(p.getProperty( "MM.ProvUserPwd", "notSet" ));
  68                 defaultProducerPort = p.getProperty( "MR.SourceReplicationPort", "9092");
  69                 defaultConsumerPort = p.getProperty( "MR.TargetReplicationPort", "2181");
  70                 centralFqdn = p.getProperty("MR.CentralCname", "notSet");
  71                 maxTopicsPerMM = Integer.valueOf( p.getProperty( "MaxTopicsPerMM", "5"));
  72                 mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
  73         }
  74
  75         // will create a MM on MMagent if needed
  76         // will update the MMagent whitelist with all topics for this MM
  77         public MirrorMaker updateMirrorMaker( MirrorMaker mm ) {
  78                 logger.info( "updateMirrorMaker");
  79
  80                 prov = new MrTopicConnection( provUser, provUserPwd );
  81
  82                 DmaapService dmaap = new DmaapService();
  83                 MR_ClusterService clusters = new MR_ClusterService();
  84                 MR_Cluster target_cluster = null;
  85                 String override = null;
  86
  87                 if ( ! mmPerMR ) {
  88                         // in ECOMP, MM Agent is only deployed at central, so this case is needed for backwards compatibility
  89                         //  we use a cname for the central MR cluster that is active, and provision on agent topic on that target
  90                         // but only send 1 message so MM Agents can read it relying on kafka delivery
  91                         for( MR_Cluster cluster: clusters.getCentralClusters() ) {
  92
  93                                 target_cluster = cluster;
  94                                 override = centralFqdn;
  95                                 // we only want to send one message even if there are multiple central clusters
  96                                 break;
  97
  98                         }
  99                 } else {
 100                         // In ONAP deployment architecture, the MM Agent is deployed with each target MR
 101                         target_cluster = clusters.getMr_ClusterByFQDN(mm.getTargetCluster());
 102                         override = null;
 103                 }
 104
 105                 prov.makeTopicConnection(target_cluster, dmaap.getBridgeAdminFqtn(), override  );
 106                 ApiError resp = prov.doPostMessage(mm.createMirrorMaker( defaultConsumerPort, defaultProducerPort ));
 107                 if ( ! resp.is2xx() ) {
 108
 109                         errorLogger.error( DmaapbcLogMessageEnum.MM_PUBLISH_ERROR, "create MM", Integer.toString(resp.getCode()), resp.getMessage());
 110                         mm.setStatus(DmaapObject_Status.INVALID);
 111                 } else {
 112                         prov.makeTopicConnection(target_cluster, dmaap.getBridgeAdminFqtn(), override );
 113                         resp = prov.doPostMessage(mm.getWhitelistUpdateJSON());
 114                         if ( ! resp.is2xx()) {
 115                                 errorLogger.error( DmaapbcLogMessageEnum.MM_PUBLISH_ERROR,"MR Bridge", Integer.toString(resp.getCode()), resp.getMessage());
 116                                 mm.setStatus(DmaapObject_Status.INVALID);
 117                         } else {
 118                                 mm.setStatus(DmaapObject_Status.VALID);
 119                         }
 120                 }
 121
 122                 mm.setLastMod();
 123                 return mirrors.put( mm.getMmName(), mm);
 124         }
 125         public MirrorMaker getMirrorMaker( String part1, String part2, int index ) {
 126                 String targetPart;
 127
 128                 // original mm names did not have any index, so leave off index 0 for
 129                 // backwards compatibility
 130                 if ( index == 0 ) {
 131                         targetPart = part2;
 132                 } else {
 133                         targetPart = part2 + "_" + index;
 134                 }
 135                 logger.info( "getMirrorMaker using " + part1 + " and " + targetPart );
 136                 return mirrors.get(MirrorMaker.genKey(part1, targetPart));
 137         }
 138         public MirrorMaker getMirrorMaker( String part1, String part2 ) {
 139                 logger.info( "getMirrorMaker using " + part1 + " and " + part2 );
 140                 return mirrors.get(MirrorMaker.genKey(part1, part2));
 141         }
 142         public MirrorMaker getMirrorMaker( String key ) {
 143                 logger.info( "getMirrorMaker using " + key);
 144                 return mirrors.get(key);
 145         }
 146
 147
 148         public void delMirrorMaker( MirrorMaker mm ) {
 149                 logger.info( "delMirrorMaker");
 150                 mirrors.remove(mm.getMmName());
 151         }
 152
 153         // TODO: this should probably return sequential values or get replaced by the MM client API
 154         // but it should be sufficient for initial 1610 development
 155         public static String genTransactionId() {
 156                 RandomInteger ri = new RandomInteger(100000);
 157             int randomInt = ri.next();
 158             return Integer.toString(randomInt);
 159         }
 160         public List<String> getAllMirrorMakers() {
 161                 List<String> ret = new ArrayList<String>();
 162                 for( String key: mirrors.keySet()) {
 163                         ret.add( key );
 164                 }
 165
 166                 return ret;
 167         }
 168
 169         public MirrorMaker getNextMM( String source, String target, String fqtn ) {
 170                 int i = 0;
 171                 MirrorMaker mm = null;
 172                 while( mm == null ) {
 173
 174                         mm = this.getMirrorMaker( source, target, i);
 175                         if ( mm == null ) {
 176                                 mm = new MirrorMaker(source, target, i);
 177                         }
 178                         if ( mm.getTopics().contains(fqtn) ) {
 179                                 break;
 180                         }
 181                         if ( mm.getTopicCount() >= maxTopicsPerMM ) {
 182                                 logger.info( "getNextMM: MM " + mm.getMmName() + " has " + mm.getTopicCount() + " topics.  Moving to next MM");
 183                                 i++;
 184                                 mm = null;
 185                         }
 186                 }
 187
 188
 189                 return mm;
 190         }
 191
 192         public MirrorMaker splitMM( MirrorMaker orig ) {
 193
 194                 String source = orig.getSourceCluster();
 195                 String target = orig.getTargetCluster();
 196
 197
 198                 ArrayList<String> whitelist = orig.getTopics();
 199                 while( whitelist.size() > maxTopicsPerMM ) {
 200
 201                         int last = whitelist.size() - 1;
 202                         String topic = whitelist.get(last);
 203                         whitelist.remove(last);
 204                         MirrorMaker mm = this.getNextMM( source, target, "aValueThatShouldNotMatchAnything" );
 205                         mm.addTopic(topic);
 206                         this.updateMirrorMaker(mm);
 207                 }
 208
 209                 orig.setTopics(whitelist);
 210
 211                 return orig;
 212
 213         }
 214
 215 }