src/main/java/org/onap/dmaap/dmf/mr/beans/DMaaPKafkaMetaBroker.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 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  *        http://www.apache.org/licenses/LICENSE-2.0
  11 *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package org.onap.dmaap.dmf.mr.beans;
  23
  24 import java.util.Arrays;
  25 import java.util.LinkedList;
  26 import java.util.List;
  27 import java.util.Properties;
  28 import java.util.Set;
  29 import java.util.TreeSet;
  30 import java.util.concurrent.ExecutionException;
  31
  32 import org.I0Itec.zkclient.ZkClient;
  33 import org.I0Itec.zkclient.exception.ZkNoNodeException;
  34 import org.apache.kafka.clients.admin.AdminClient;
  35 import org.apache.kafka.clients.admin.AdminClientConfig;
  36 import org.apache.kafka.clients.admin.CreateTopicsResult;
  37 import org.apache.kafka.clients.admin.NewTopic;
  38 import org.apache.kafka.common.KafkaFuture;
  39 import org.json.JSONObject;
  40 import org.json.JSONArray;
  41 import org.springframework.beans.factory.annotation.Qualifier;
  42 import org.springframework.util.StringUtils;
  43 import org.onap.dmaap.dmf.mr.CambriaApiException;
  44 import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
  45 import org.onap.dmaap.dmf.mr.metabroker.Broker1;
  46 import org.onap.dmaap.dmf.mr.metabroker.Topic;
  47 import org.onap.dmaap.dmf.mr.utils.ConfigurationReader;
  48 import org.onap.dmaap.dmf.mr.utils.Utils;
  49 import com.att.eelf.configuration.EELFLogger;
  50 import com.att.eelf.configuration.EELFManager;
  51
  52 import com.att.nsa.configs.ConfigDb;
  53 import com.att.nsa.configs.ConfigDbException;
  54 import com.att.nsa.configs.ConfigPath;
  55 import com.att.nsa.drumlin.service.standards.HttpStatusCodes;
  56 import com.att.nsa.drumlin.till.nv.rrNvReadable;
  57 import com.att.nsa.security.NsaAcl;
  58 import com.att.nsa.security.NsaAclUtils;
  59 import com.att.nsa.security.NsaApiKey;
  60
  61
  62 /**
  63  * class performing all topic operations
  64  *
  65  * @author anowarul.islam
  66  *
  67  */
  68 //@Component
  69 public class DMaaPKafkaMetaBroker implements Broker1 {
  70
  71         public DMaaPKafkaMetaBroker() {
  72                 fZk = null;
  73                 fCambriaConfig = null;
  74                 fBaseTopicData = null;
  75                 final Properties props = new Properties ();
  76                 String fkafkaBrokers = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
  77                                 "kafka.metadata.broker.list");
  78                 if (StringUtils.isEmpty(fkafkaBrokers)) {
  79
  80                         fkafkaBrokers = "localhost:9092";
  81                 }
  82
  83              props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, fkafkaBrokers );
  84              if(Utils.isCadiEnabled()){
  85              props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='"+Utils.getKafkaproperty()+"';");
  86                  props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
  87              props.put("sasl.mechanism", "PLAIN");
  88              }
  89
  90              fKafkaAdminClient=AdminClient.create ( props );
  91
  92         }
  93
  94         private static final EELFLogger log = EELFManager.getInstance().getLogger(ConfigurationReader.class);
  95         private final AdminClient fKafkaAdminClient;
  96
  97
  98
  99         /**
 100          * DMaaPKafkaMetaBroker constructor initializing
 101          *
 102          * @param settings
 103          * @param zk
 104          * @param configDb
 105          */
 106         public DMaaPKafkaMetaBroker(@Qualifier("propertyReader") rrNvReadable settings,
 107                         @Qualifier("dMaaPZkClient") ZkClient zk, @Qualifier("dMaaPZkConfigDb") ConfigDb configDb) {
 108                 fZk = zk;
 109                 fCambriaConfig = configDb;
 110                 fBaseTopicData = configDb.parse("/topics");
 111                 final Properties props = new Properties ();
 112                 String fkafkaBrokers = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
 113                                 "kafka.metadata.broker.list");
 114                 if (null == fkafkaBrokers) {
 115
 116                         fkafkaBrokers = "localhost:9092";
 117                 }
 118
 119                  if(Utils.isCadiEnabled()){
 120                  props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='"+Utils.getKafkaproperty()+"';");
 121                  props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
 122              props.put("sasl.mechanism", "PLAIN");
 123                  }
 124              props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, fkafkaBrokers );
 125
 126              fKafkaAdminClient=AdminClient.create ( props );
 127
 128
 129
 130         }
 131
 132         public DMaaPKafkaMetaBroker( rrNvReadable settings,
 133                         ZkClient zk,  ConfigDb configDb,AdminClient client) {
 134
 135                 fZk = zk;
 136                 fCambriaConfig = configDb;
 137                 fBaseTopicData = configDb.parse("/topics");
 138             fKafkaAdminClient= client;
 139
 140
 141
 142         }
 143
 144         @Override
 145         public List<Topic> getAllTopics() throws ConfigDbException {
 146                 log.info("Retrieving list of all the topics.");
 147                 final LinkedList<Topic> result = new LinkedList<>();
 148                 try {
 149                         log.info("Retrieving all topics from root: " + zkTopicsRoot);
 150                         final List<String> topics = fZk.getChildren(zkTopicsRoot);
 151                         for (String topic : topics) {
 152                                 result.add(new KafkaTopic(topic, fCambriaConfig, fBaseTopicData));
 153                         }
 154                         JSONObject dataObj = new JSONObject();
 155                         dataObj.put("topics", new JSONObject());
 156
 157                         for (String topic : topics) {
 158                                 dataObj.getJSONObject("topics").put(topic, new JSONObject());
 159                         }
 160                 } catch (ZkNoNodeException excp) {
 161                         // very fresh kafka doesn't have any topics or a topics node
 162                         log.error("ZK doesn't have a Kakfa topics node at " + zkTopicsRoot, excp);
 163                 }
 164                 return result;
 165         }
 166
 167         @Override
 168         public Topic getTopic(String topic) throws ConfigDbException {
 169                 if (fZk.exists(zkTopicsRoot + "/" + topic)) {
 170                         return getKafkaTopicConfig(fCambriaConfig, fBaseTopicData, topic);
 171                 }
 172                 // else: no such topic in kafka
 173                 return null;
 174         }
 175
 176         /**
 177          * static method get KafkaTopic object
 178          *
 179          * @param db
 180          * @param base
 181          * @param topic
 182          * @return
 183          * @throws ConfigDbException
 184          */
 185         public static KafkaTopic getKafkaTopicConfig(ConfigDb db, ConfigPath base, String topic) throws ConfigDbException {
 186                 return new KafkaTopic(topic, db, base);
 187         }
 188
 189         /**
 190          * creating topic
 191          */
 192         @Override
 193         public Topic createTopic(String topic, String desc, String ownerApiKey, int partitions, int replicas,
 194                         boolean transactionEnabled) throws TopicExistsException, CambriaApiException,ConfigDbException {
 195                 log.info("Creating topic: " + topic);
 196                 try {
 197                         log.info("Check if topic [" + topic + "] exist.");
 198                         // first check for existence "our way"
 199                         final Topic t = getTopic(topic);
 200                         if (t != null) {
 201                                 log.info("Could not create topic [" + topic + "]. Topic Already exists.");
 202                                 throw new TopicExistsException("Could not create topic [" + topic + "]. Topic Alreay exists.");
 203                         }
 204                 } catch (ConfigDbException e1) {
 205                         log.error("Topic [" + topic + "] could not be created. Couldn't check topic data in config db.", e1);
 206                         throw new CambriaApiException(HttpStatusCodes.k503_serviceUnavailable,
 207                                         "Couldn't check topic data in config db.");
 208                 }
 209
 210                 // we only allow 3 replicas. (If we don't test this, we get weird
 211                 // results from the cluster,
 212                 // so explicit test and fail.)
 213                 if (replicas < 1 || replicas > 3) {
 214                         log.info("Topic [" + topic + "] could not be created. The replica count must be between 1 and 3.");
 215                         throw new CambriaApiException(HttpStatusCodes.k400_badRequest,
 216                                         "The replica count must be between 1 and 3.");
 217                 }
 218                 if (partitions < 1) {
 219                         log.info("Topic [" + topic + "] could not be created. The partition count must be at least 1.");
 220                         throw new CambriaApiException(HttpStatusCodes.k400_badRequest, "The partition count must be at least 1.");
 221                 }
 222
 223                 // create via kafka
 224
 225         try {
 226             final NewTopic topicRequest =
 227                     new NewTopic(topic, partitions, (short)replicas);
 228             final CreateTopicsResult ctr =
 229                     fKafkaAdminClient.createTopics(Arrays.asList(topicRequest));
 230             final KafkaFuture<Void> ctrResult = ctr.all();
 231             ctrResult.get();
 232             // underlying Kafka topic created. now setup our API info
 233             return createTopicEntry(topic, desc, ownerApiKey, transactionEnabled);
 234         } catch (InterruptedException e) {
 235             log.warn("Execution of describeTopics timed out.");
 236             throw new ConfigDbException(e);
 237         } catch (ExecutionException e) {
 238             log.warn("Execution of describeTopics failed: " + e.getCause().getMessage(), e);
 239             throw new ConfigDbException(e.getCause());
 240         }
 241
 242         }
 243
 244         @Override
 245         public void deleteTopic(String topic) throws CambriaApiException, TopicExistsException,ConfigDbException {
 246                 log.info("Deleting topic: " + topic);
 247                 try {
 248                         log.info("Loading zookeeper client for topic deletion.");
 249                                         // topic creation. (Otherwise, the topic is only partially created
 250                         // in ZK.)
 251
 252
 253                         fKafkaAdminClient.deleteTopics(Arrays.asList(topic));
 254                         log.info("Zookeeper client loaded successfully. Deleting topic.");
 255
 256                 } catch (Exception e) {
 257                         log.error("Failed to delete topic [" + topic + "]. " + e.getMessage(), e);
 258                         throw new ConfigDbException(e);
 259                 }  finally {
 260                         log.info("Closing zookeeper connection.");
 261                 }
 262         }
 263
 264         private final ZkClient fZk;
 265         private final ConfigDb fCambriaConfig;
 266         private final ConfigPath fBaseTopicData;
 267
 268         private static final String zkTopicsRoot = "/brokers/topics";
 269         private static final JSONObject kEmptyAcl = new JSONObject();
 270
 271         /**
 272          * method Providing KafkaTopic Object associated with owner and
 273          * transactionenabled or not
 274          *
 275          * @param name
 276          * @param desc
 277          * @param owner
 278          * @param transactionEnabled
 279          * @return
 280          * @throws ConfigDbException
 281          */
 282         public KafkaTopic createTopicEntry(String name, String desc, String owner, boolean transactionEnabled)
 283                         throws ConfigDbException {
 284                 return createTopicEntry(fCambriaConfig, fBaseTopicData, name, desc, owner, transactionEnabled);
 285         }
 286
 287         /**
 288          * static method giving kafka topic object
 289          *
 290          * @param db
 291          * @param basePath
 292          * @param name
 293          * @param desc
 294          * @param owner
 295          * @param transactionEnabled
 296          * @return
 297          * @throws ConfigDbException
 298          */
 299         public static KafkaTopic createTopicEntry(ConfigDb db, ConfigPath basePath, String name, String desc, String owner,
 300                         boolean transactionEnabled) throws ConfigDbException {
 301                 final JSONObject o = new JSONObject();
 302                 o.put("owner", owner);
 303                 o.put("description", desc);
 304                 o.put("txenabled", transactionEnabled);
 305                 db.store(basePath.getChild(name), o.toString());
 306                 return new KafkaTopic(name, db, basePath);
 307         }
 308
 309         /**
 310          * class performing all user opearation like user is eligible to read,
 311          * write. permitting a user to write and read,
 312          *
 313          * @author anowarul.islam
 314          *
 315          */
 316         public static class KafkaTopic implements Topic {
 317                 /**
 318                  * constructor initializes
 319                  *
 320                  * @param name
 321                  * @param configdb
 322                  * @param baseTopic
 323                  * @throws ConfigDbException
 324                  */
 325                 public KafkaTopic(String name, ConfigDb configdb, ConfigPath baseTopic) throws ConfigDbException {
 326                         fName = name;
 327                         fConfigDb = configdb;
 328                         fBaseTopicData = baseTopic;
 329
 330                         String data = fConfigDb.load(fBaseTopicData.getChild(fName));
 331                         if (data == null) {
 332                                 data = "{}";
 333                         }
 334
 335                         final JSONObject o = new JSONObject(data);
 336                         fOwner = o.optString("owner", "");
 337                         fDesc = o.optString("description", "");
 338                         fTransactionEnabled = o.optBoolean("txenabled", false);// default
 339                                                                                                                                         // value is
 340                                                                                                                                         // false
 341                         // if this topic has an owner, it needs both read/write ACLs. If there's no
 342                                                 // owner (or it's empty), null is okay -- this is for existing or implicitly
 343                                                 // created topics.
 344                                                 JSONObject readers = o.optJSONObject ( "readers" );
 345                                                 if ( readers == null && fOwner.length () > 0 )
 346                                                 {
 347                                                     readers = kEmptyAcl;
 348                                                 }
 349                                                 fReaders =  fromJson ( readers );
 350
 351                                                 JSONObject writers = o.optJSONObject ( "writers" );
 352                                                 if ( writers == null && fOwner.length () > 0 )
 353                                                 {
 354                                                     writers = kEmptyAcl;
 355                                                 }
 356                                                 fWriters = fromJson ( writers );
 357                 }
 358
 359                 private NsaAcl fromJson(JSONObject o) {
 360                         NsaAcl acl = new NsaAcl();
 361                         if (o != null) {
 362                                 JSONArray a = o.optJSONArray("allowed");
 363                                 if (a != null) {
 364                                         for (int i = 0; i < a.length(); ++i) {
 365                                                 String user = a.getString(i);
 366                                                 acl.add(user);
 367                                         }
 368                                 }
 369                         }
 370                         return acl;
 371                 }
 372
 373                 @Override
 374                 public String getName() {
 375                         return fName;
 376                 }
 377
 378                 @Override
 379                 public String getOwner() {
 380                         return fOwner;
 381                 }
 382
 383                 @Override
 384                 public String getDescription() {
 385                         return fDesc;
 386                 }
 387
 388                 @Override
 389                 public NsaAcl getReaderAcl() {
 390                         return fReaders;
 391                 }
 392
 393                 @Override
 394                 public NsaAcl getWriterAcl() {
 395                         return fWriters;
 396                 }
 397
 398                 @Override
 399                 public void checkUserRead(NsaApiKey user) throws AccessDeniedException  {
 400                         NsaAclUtils.checkUserAccess ( fOwner, getReaderAcl(), user );
 401                 }
 402
 403                 @Override
 404                 public void checkUserWrite(NsaApiKey user) throws AccessDeniedException  {
 405                         NsaAclUtils.checkUserAccess ( fOwner, getWriterAcl(), user );
 406                 }
 407
 408                 @Override
 409                 public void permitWritesFromUser(String pubId, NsaApiKey asUser)
 410                                 throws ConfigDbException, AccessDeniedException {
 411                         updateAcl(asUser, false, true, pubId);
 412                 }
 413
 414                 @Override
 415                 public void denyWritesFromUser(String pubId, NsaApiKey asUser) throws ConfigDbException, AccessDeniedException {
 416                         updateAcl(asUser, false, false, pubId);
 417                 }
 418
 419                 @Override
 420                 public void permitReadsByUser(String consumerId, NsaApiKey asUser)
 421                                 throws ConfigDbException, AccessDeniedException {
 422                         updateAcl(asUser, true, true, consumerId);
 423                 }
 424
 425                 @Override
 426                 public void denyReadsByUser(String consumerId, NsaApiKey asUser)
 427                                 throws ConfigDbException, AccessDeniedException {
 428                         updateAcl(asUser, true, false, consumerId);
 429                 }
 430
 431                 private void updateAcl(NsaApiKey asUser, boolean reader, boolean add, String key)
 432                                 throws ConfigDbException, AccessDeniedException{
 433                         try
 434                         {
 435                                 final NsaAcl acl = NsaAclUtils.updateAcl ( this, asUser, key, reader, add );
 436
 437                                 // we have to assume we have current data, or load it again. for the expected use
 438                                 // case, assuming we can overwrite the data is fine.
 439                                 final JSONObject o = new JSONObject ();
 440                                 o.put ( "owner", fOwner );
 441                                 o.put ( "readers", safeSerialize ( reader ? acl : fReaders ) );
 442                                 o.put ( "writers", safeSerialize ( reader ? fWriters : acl ) );
 443                                 fConfigDb.store ( fBaseTopicData.getChild ( fName ), o.toString () );
 444
 445                                 log.info ( "ACL_UPDATE: " + asUser.getKey () + " " + ( add ? "added" : "removed" ) + ( reader?"subscriber":"publisher" ) + " " + key + " on " + fName );
 446
 447                         }
 448                         catch ( ConfigDbException | AccessDeniedException x )
 449                         {
 450                                 throw x;
 451                         }
 452
 453                 }
 454
 455                 private JSONObject safeSerialize(NsaAcl acl) {
 456                         return acl == null ? null : acl.serialize();
 457                 }
 458
 459                 private final String fName;
 460                 private final ConfigDb fConfigDb;
 461                 private final ConfigPath fBaseTopicData;
 462                 private final String fOwner;
 463                 private final String fDesc;
 464                 private final NsaAcl fReaders;
 465                 private final NsaAcl fWriters;
 466                 private boolean fTransactionEnabled;
 467
 468                 public boolean isTransactionEnabled() {
 469                         return fTransactionEnabled;
 470                 }
 471
 472                 @Override
 473                 public Set<String> getOwners() {
 474                         final TreeSet<String> owners = new TreeSet<>();
 475                         owners.add ( fOwner );
 476                         return owners;
 477                 }
 478         }
 479
 480 }