2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
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
13 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
23 package org.onap.dmaap.dbcapi.service;
25 import java.util.ArrayList;
26 import java.util.Collection;
27 import java.util.Iterator;
28 import java.util.List;
32 import javax.ws.rs.core.Response.Status;
34 import org.onap.dmaap.dbcapi.aaf.AafNamespace;
35 import org.onap.dmaap.dbcapi.aaf.AafRole;
36 import org.onap.dmaap.dbcapi.aaf.AafService;
37 import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
38 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
39 import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
40 import org.onap.dmaap.dbcapi.database.DatabaseClass;
41 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
42 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
43 import org.onap.dmaap.dbcapi.model.ApiError;
44 import org.onap.dmaap.dbcapi.model.DcaeLocation;
45 import org.onap.dmaap.dbcapi.model.Dmaap;
46 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
47 import org.onap.dmaap.dbcapi.model.MR_Client;
48 import org.onap.dmaap.dbcapi.model.MR_Cluster;
49 import org.onap.dmaap.dbcapi.model.MirrorMaker;
50 import org.onap.dmaap.dbcapi.model.ReplicationType;
51 import org.onap.dmaap.dbcapi.model.Topic;
52 import org.onap.dmaap.dbcapi.util.DmaapConfig;
53 import org.onap.dmaap.dbcapi.util.Fqdn;
54 import org.onap.dmaap.dbcapi.util.Graph;
56 public class TopicService extends BaseLoggingClass {
60 // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
61 private static String defaultGlobalMrHost;
63 private Map<String, Topic> mr_topics = DatabaseClass.getTopics();
65 private static DmaapService dmaapSvc = new DmaapService();
66 private MR_ClientService clientService = new MR_ClientService();
67 private MR_ClusterService clusters = new MR_ClusterService();
68 private DcaeLocationService locations = new DcaeLocationService();
69 private MirrorMakerService bridge = new MirrorMakerService();
71 private static String centralCname;
72 private static boolean createTopicRoles;
75 public TopicService(){
76 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
77 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
78 centralCname = p.getProperty("MR.CentralCname");
79 createTopicRoles = "true".equalsIgnoreCase(p.getProperty("aaf.CreateTopicRoles", "true"));
82 logger.info( "TopicService properties: CentralCname=" + centralCname +
83 " defaultGlobarlMrHost=" + defaultGlobalMrHost +
84 " createTopicRoles=" + createTopicRoles );
87 public Map<String, Topic> getTopics() {
91 public List<Topic> getAllTopics() {
92 return getAllTopics( true );
94 public List<Topic> getAllTopicsWithoutClients() {
95 return getAllTopics(false);
98 private List<Topic> getAllTopics( Boolean withClients ) {
99 ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
101 for( Topic topic: topics ) {
102 topic.setClients( clientService.getAllMrClients(topic.getFqtn()));
109 public Topic getTopic( String key, ApiError apiError ) {
110 logger.info( "getTopic: key=" + key);
111 Topic t = mr_topics.get( key );
113 apiError.setCode(Status.NOT_FOUND.getStatusCode());
114 apiError.setFields( "fqtn");
115 apiError.setMessage("topic with fqtn " + key + " not found");
118 t.setClients( clientService.getAllMrClients( key ));
119 apiError.setCode(Status.OK.getStatusCode());
123 private void aafTopicSetup(Topic topic, ApiError err ) {
125 String nsr = dmaapSvc.getDmaap().getTopicNsRoot();
128 err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)" );
129 err.setFields("topicNsRoot");
133 // establish AAF Connection using TopicMgr identity
134 AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
136 AafRole pubRole = null;
137 AafRole subRole = null;
139 // creating Topic Roles was not an original feature.
140 // For backwards compatibility, only do this if the feature is enabled.
141 // Also, if the namespace of the topic is a foreign namespace, (i.e. not the same as our root ns)
142 // then we likely don't have permission to create sub-ns and Roles so don't try.
143 if ( createTopicRoles && topic.getFqtn().startsWith(nsr)) {
144 // create AAF namespace for this topic
145 AafNamespace ns = new AafNamespace( topic.getFqtn(), aaf.getIdentity());
147 int rc = aaf.addNamespace( ns );
148 if ( rc != 201 && rc != 409 ) {
150 err.setMessage("Unexpected response from AAF:" + rc );
151 err.setFields("namespace:" + topic.getFqtn() + " identity="+ aaf.getIdentity());
156 // create AAF Roles for MR clients of this topic
157 String rn = "publisher";
158 pubRole = new AafRole( topic.getFqtn(), rn );
159 int rc = aaf.addRole( pubRole );
160 if ( rc != 201 && rc != 409 ) {
162 err.setMessage("Unexpected response from AAF:" + rc );
163 err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
166 topic.setPublisherRole( pubRole.getFullyQualifiedRole() );
169 subRole = new AafRole( topic.getFqtn(), rn );
170 rc = aaf.addRole( subRole );
171 if ( rc != 201 && rc != 409 ) {
173 err.setMessage("Unexpected response from AAF:" + rc );
174 err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
177 topic.setSubscriberRole( subRole.getFullyQualifiedRole() );
180 // create AAF perms checked by MR
181 String instance = ":topic." + topic.getFqtn();
182 String[] actions = { "pub", "sub", "view" };
183 String t = dmaapSvc.getTopicPerm();
184 for ( String action : actions ){
185 DmaapPerm perm = new DmaapPerm( t, instance, action );
186 int rc = aaf.addPerm( perm );
187 if ( rc != 201 && rc != 409 ) {
189 err.setMessage("Unexpected response from AAF:" + rc );
190 err.setFields("t="+t + " instance="+ instance + " action="+ action);
193 if ( createTopicRoles ) {
194 // Grant perms to our default Roles
195 if ( action.equals( "pub") || action.equals( "view") ) {
196 DmaapGrant g = new DmaapGrant( perm, pubRole.getFullyQualifiedRole() );
197 rc = aaf.addGrant( g );
198 if ( rc != 201 && rc != 409 ) {
200 err.setMessage( "Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole() );
201 logger.warn( err.getMessage());
205 if ( action.equals( "sub") || action.equals( "view") ) {
206 DmaapGrant g = new DmaapGrant( perm, subRole.getFullyQualifiedRole() );
207 rc = aaf.addGrant( g );
208 if ( rc != 201 && rc != 409 ) {
210 err.setMessage( "Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole() );
211 logger.warn( err.getMessage());
220 public Topic addTopic( Topic topic, ApiError err, Boolean useExisting ) {
221 logger.info( "Entry: addTopic");
222 logger.info( "Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle() );
223 String nFqtn = topic.genFqtn();
224 logger.info( "FQTN=" + nFqtn );
225 Topic pTopic = getTopic( nFqtn, err );
226 if ( pTopic != null ) {
227 String t = "topic already exists: " + nFqtn;
230 err.setCode(Status.OK.getStatusCode());
234 err.setFields( "fqtn");
235 err.setCode(Status.CONFLICT.getStatusCode());
238 err.reset(); // err filled with NOT_FOUND is expected case, but don't want to litter...
240 topic.setFqtn( nFqtn );
242 aafTopicSetup( topic, err );
243 if ( err.getCode() >= 400 ) {
247 if ( topic.getReplicationCase().involvesGlobal() ) {
248 if ( topic.getGlobalMrURL() == null ) {
249 topic.setGlobalMrURL(defaultGlobalMrHost);
251 if ( ! Fqdn.isValid( topic.getGlobalMrURL())) {
252 logger.error( "GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
253 topic.setStatus( DmaapObject_Status.INVALID);
255 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL() );
256 err.setFields("globalMrURL");
263 if ( topic.getNumClients() > 0 ) {
264 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
267 ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
268 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
269 MR_Client c = it.next();
271 logger.info( "c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
272 MR_Client nc = new MR_Client( c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
273 nc.setFqtn(topic.getFqtn());
274 nc.setClientIdentity( c.getClientIdentity());
275 logger.info( "nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
276 clients2.add( clientService.addMr_Client(nc, topic, err));
277 if ( ! err.is2xx()) {
282 topic.setClients(clients2);
285 Topic ntopic = checkForBridge( topic, err );
286 if ( ntopic == null ) {
287 topic.setStatus( DmaapObject_Status.INVALID);
288 if ( ! err.is2xx()) {
294 mr_topics.put( nFqtn, ntopic );
296 err.setCode(Status.OK.getStatusCode());
301 public Topic updateTopic( Topic topic, ApiError err ) {
302 logger.info( "Entry: updateTopic");
303 if ( topic.getFqtn().isEmpty()) {
306 Topic ntopic = checkForBridge( topic, err );
307 if ( ntopic == null ) {
308 topic.setStatus( DmaapObject_Status.INVALID);
309 if ( ! err.is2xx() ) {
314 mr_topics.put( ntopic.getFqtn(), ntopic );
316 err.setCode(Status.OK.getStatusCode());
320 public Topic removeTopic( String pubId, ApiError apiError ) {
321 Topic topic = mr_topics.get(pubId);
322 if ( topic == null ) {
323 apiError.setCode(Status.NOT_FOUND.getStatusCode());
324 apiError.setMessage("Topic " + pubId + " does not exist");
325 apiError.setFields("fqtn");
328 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients( pubId ));
329 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
330 MR_Client c = it.next();
333 clientService.removeMr_Client(c.getMrClientId(), false, apiError);
334 if ( ! apiError.is2xx()) {
338 apiError.setCode(Status.OK.getStatusCode());
339 return mr_topics.remove(pubId);
341 public static ApiError setBridgeClientPerms( MR_Cluster node ) {
342 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
343 String mmProvRole = p.getProperty("MM.ProvRole");
344 String mmAgentRole = p.getProperty("MM.AgentRole");
345 String[] Roles = { mmProvRole, mmAgentRole };
346 String[] actions = { "view", "pub", "sub" };
347 Topic bridgeAdminTopic = new Topic().init();
348 bridgeAdminTopic.setTopicName( dmaapSvc.getBridgeAdminFqtn() );
349 bridgeAdminTopic.setTopicDescription( "RESERVED topic for MirroMaker Provisioning");
350 bridgeAdminTopic.setOwner( "DBC" );
352 ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
353 for( String role: Roles ) {
354 MR_Client client = new MR_Client();
355 client.setAction(actions);
356 client.setClientRole(role);
357 client.setDcaeLocationName( node.getDcaeLocationName());
358 clients.add( client );
360 bridgeAdminTopic.setClients(clients);
362 TopicService ts = new TopicService();
363 ApiError err = new ApiError();
364 ts.addTopic(bridgeAdminTopic, err, true);
366 if ( err.is2xx() || err.getCode() == 409 ){
371 errorLogger.error( DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
376 public Topic checkForBridge( Topic topic, ApiError err ) {
378 if ( topic.getReplicationCase() == ReplicationType.REPLICATION_NONE ) {
379 topic.setStatus( DmaapObject_Status.VALID);
383 boolean anythingWrong = false;
385 Set<String> groups = clusters.getGroups();
386 for ( String g : groups ) {
387 anythingWrong |= buildBridge( topic, err, g );
389 if ( anythingWrong ) {
390 topic.setStatus( DmaapObject_Status.INVALID);
391 if ( ! err.is2xx() ) {
395 topic.setStatus( DmaapObject_Status.VALID);
400 private boolean buildBridge( Topic topic, ApiError err, String group ) {
402 boolean anythingWrong = false;
404 if ( group == null || group.isEmpty() ) {
405 graph = new Graph( topic.getClients(), true );
407 graph = new Graph( topic.getClients(), true, group );
409 MR_Cluster groupCentralCluster = null;
411 if ( graph.isEmpty() ) {
413 } else if ( group == null && topic.getReplicationCase().involvesFQDN() ) {
415 } else if ( ! graph.hasCentral() ) {
416 logger.warn( "Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
419 groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
421 Collection<String> clientLocations = graph.getKeys();
422 for( String loc : clientLocations ) {
423 logger.info( "loc=" + loc );
424 DcaeLocation location = locations.getDcaeLocation(loc);
425 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
426 logger.info( "cluster=" + cluster );
430 String source = null;
431 String target = null;
434 * Provision Edge to Central bridges...
436 if ( ! location.isCentral() && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
437 switch( topic.getReplicationCase() ) {
438 case REPLICATION_EDGE_TO_CENTRAL:
439 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL: // NOTE: this is for E2C portion only
440 source = cluster.getFqdn();
441 target = centralCname;
443 case REPLICATION_CENTRAL_TO_EDGE:
444 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for C2E portion only
445 source = centralCname;
446 target = cluster.getFqdn();
448 case REPLICATION_CENTRAL_TO_GLOBAL:
449 case REPLICATION_GLOBAL_TO_CENTRAL:
450 case REPLICATION_FQDN_TO_GLOBAL:
451 case REPLICATION_GLOBAL_TO_FQDN:
454 case REPLICATION_EDGE_TO_FQDN:
455 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2C portion only
456 source = cluster.getFqdn();
457 target = groupCentralCluster.getFqdn();
459 case REPLICATION_FQDN_TO_EDGE:
460 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for F2E portion only
461 source = groupCentralCluster.getFqdn();
462 target = cluster.getFqdn();
466 logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
467 anythingWrong = true;
469 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
470 + topic.getReplicationCase() );
471 err.setMessage("Unexpected value for ReplicationType");
475 } else if ( location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
477 * Provision Central to Global bridges
479 switch( topic.getReplicationCase() ) {
481 case REPLICATION_CENTRAL_TO_GLOBAL:
482 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
483 source = centralCname;
484 target = topic.getGlobalMrURL();
486 case REPLICATION_GLOBAL_TO_CENTRAL:
487 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for G2C portion only
488 source = topic.getGlobalMrURL();
489 target = centralCname;
492 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2F portion only
493 source = groupCentralCluster.getFqdn();
494 target = topic.getGlobalMrURL();
497 case REPLICATION_FQDN_TO_GLOBAL:
498 source = groupCentralCluster.getFqdn();
499 target = topic.getGlobalMrURL();
502 case REPLICATION_GLOBAL_TO_FQDN:
503 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for G2F portion only
504 source = topic.getGlobalMrURL();
505 target = groupCentralCluster.getFqdn();
508 case REPLICATION_FQDN_TO_EDGE:
509 case REPLICATION_EDGE_TO_FQDN:
510 case REPLICATION_EDGE_TO_CENTRAL:
511 case REPLICATION_CENTRAL_TO_EDGE:
514 logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
515 anythingWrong = true;
517 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
518 + topic.getReplicationCase() );
519 err.setMessage("Unexpected value for ReplicationType");
523 logger.warn( "dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
524 anythingWrong = true;
527 if ( source != null && target != null ) {
529 logger.info( "Create a MM from " + source + " to " + target );
530 MirrorMaker mm = bridge.getNextMM( source, target, topic.getFqtn());
531 mm.addTopic(topic.getFqtn());
532 bridge.updateMirrorMaker(mm);
533 } catch ( Exception ex ) {
535 err.setFields( "mirror_maker.topic");
536 err.setMessage("Unexpected condition: " + ex );
537 anythingWrong = true;
544 return anythingWrong;
550 * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
551 * This was determined automatically based on presence of edge publishers and central subscribers.
552 * The following method is a modification of that original logic, to preserve some backwards compatibility,
553 * i.e. to be used when no ReplicationType is specified.
555 public ReplicationType reviewTopic( Topic topic ) {
558 if ( topic.getNumClients() > 1 ) {
559 Graph graph = new Graph( topic.getClients(), false );
561 String centralFqdn = new String();
562 if ( graph.hasCentral() ) {
563 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
564 centralFqdn = p.getProperty("MR.CentralCname");
567 Collection<String> locations = graph.getKeys();
568 for( String loc : locations ) {
569 logger.info( "loc=" + loc );
570 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
571 if ( cluster == null ) {
572 logger.info( "No MR cluster for location " + loc );
575 if ( graph.hasCentral() && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
576 logger.info( "Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn );
577 return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
584 return ReplicationType.REPLICATION_NONE;