2 * ============LICENSE_START=======================================================
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
21 package org.onap.dmaap.dbcapi.service;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.Iterator;
26 import java.util.List;
30 import javax.ws.rs.core.Response.Status;
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;
54 public class TopicService extends BaseLoggingClass {
58 // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
59 private static String defaultGlobalMrHost;
61 private Map<String, Topic> mr_topics = DatabaseClass.getTopics();
63 private static DmaapService dmaapSvc = new DmaapService();
64 private MR_ClientService clientService = new MR_ClientService();
65 private MR_ClusterService clusters = new MR_ClusterService();
66 private DcaeLocationService locations = new DcaeLocationService();
67 private MirrorMakerService bridge = new MirrorMakerService();
69 private static String centralCname;
72 public TopicService(){
73 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
74 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
75 centralCname = p.getProperty("MR.CentralCname");
78 logger.info( "TopicService properties: CentralCname=" + centralCname +
79 " defaultGlobarlMrHost=" + defaultGlobalMrHost );
82 public Map<String, Topic> getTopics() {
86 public List<Topic> getAllTopics() {
87 return getAllTopics( true );
89 public List<Topic> getAllTopicsWithoutClients() {
90 return getAllTopics(false);
93 private List<Topic> getAllTopics( Boolean withClients ) {
94 ArrayList<Topic> topics = new ArrayList<Topic>(mr_topics.values());
96 for( Topic topic: topics ) {
97 topic.setClients( clientService.getAllMrClients(topic.getFqtn()));
104 public Topic getTopic( String key, ApiError apiError ) {
105 logger.info( "getTopic: key=" + key);
106 Topic t = mr_topics.get( key );
108 apiError.setCode(Status.NOT_FOUND.getStatusCode());
109 apiError.setFields( "fqtn");
110 apiError.setMessage("topic with fqtn " + key + " not found");
113 t.setClients( clientService.getAllMrClients( key ));
114 apiError.setCode(Status.OK.getStatusCode());
118 private void aafTopicSetup(Topic topic, ApiError err ) {
120 String t = dmaapSvc.getTopicPerm();
123 err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)" );
124 err.setFields("topicNsRoot");
128 // establish AAF Connection using TopicMgr identity
129 AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
133 // create AAF namespace for this topic
134 AafNamespace ns = new AafNamespace( topic.getFqtn(), aaf.getIdentity());
136 int rc = aaf.addNamespace( ns );
137 if ( rc != 201 && rc != 409 ) {
139 err.setMessage("Unexpected response from AAF:" + rc );
140 err.setFields("namespace:" + topic.getFqtn() + " identity="+ aaf.getIdentity());
145 // create AAF Roles for MR clients of this topic
146 String rn = "publisher";
147 AafRole pubRole = new AafRole( topic.getFqtn(), rn );
148 int rc = aaf.addRole( pubRole );
149 if ( rc != 201 && rc != 409 ) {
151 err.setMessage("Unexpected response from AAF:" + rc );
152 err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
155 topic.setPublisherRole( pubRole.getFullyQualifiedRole() );
158 AafRole subRole = new AafRole( topic.getFqtn(), rn );
159 rc = aaf.addRole( subRole );
160 if ( rc != 201 && rc != 409 ) {
162 err.setMessage("Unexpected response from AAF:" + rc );
163 err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
166 topic.setSubscriberRole( subRole.getFullyQualifiedRole() );
169 // create AAF perms checked by MR
170 String instance = ":topic." + topic.getFqtn();
171 String[] actions = { "pub", "sub", "view" };
172 for ( String action : actions ){
173 DmaapPerm perm = new DmaapPerm( t, instance, action );
174 rc = aaf.addPerm( perm );
175 if ( rc != 201 && rc != 409 ) {
177 err.setMessage("Unexpected response from AAF:" + rc );
178 err.setFields("t="+t + " instance="+ instance + " action="+ action);
181 // Grant perms to our default Roles
182 if ( action.equals( "pub") || action.equals( "view") ) {
183 DmaapGrant g = new DmaapGrant( perm, pubRole.getFullyQualifiedRole() );
184 rc = aaf.addGrant( g );
185 if ( rc != 201 && rc != 409 ) {
187 err.setMessage( "Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole() );
188 logger.warn( err.getMessage());
192 if ( action.equals( "sub") || action.equals( "view") ) {
193 DmaapGrant g = new DmaapGrant( perm, subRole.getFullyQualifiedRole() );
194 rc = aaf.addGrant( g );
195 if ( rc != 201 && rc != 409 ) {
197 err.setMessage( "Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole() );
198 logger.warn( err.getMessage());
206 public Topic addTopic( Topic topic, ApiError err, Boolean useExisting ) {
207 logger.info( "Entry: addTopic");
208 logger.info( "Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle() );
209 String nFqtn = topic.genFqtn();
210 logger.info( "FQTN=" + nFqtn );
211 Topic pTopic = getTopic( nFqtn, err );
212 if ( pTopic != null ) {
213 String t = "topic already exists: " + nFqtn;
216 err.setCode(Status.OK.getStatusCode());
220 err.setFields( "fqtn");
221 err.setCode(Status.CONFLICT.getStatusCode());
224 err.reset(); // err filled with NOT_FOUND is expected case, but don't want to litter...
226 topic.setFqtn( nFqtn );
228 aafTopicSetup( topic, err );
229 if ( err.getCode() >= 400 ) {
233 if ( topic.getReplicationCase().involvesGlobal() ) {
234 if ( topic.getGlobalMrURL() == null ) {
235 topic.setGlobalMrURL(defaultGlobalMrHost);
237 if ( ! Fqdn.isValid( topic.getGlobalMrURL())) {
238 logger.error( "GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
239 topic.setStatus( DmaapObject_Status.INVALID);
241 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL() );
242 err.setFields("globalMrURL");
249 if ( topic.getNumClients() > 0 ) {
250 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
253 ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
254 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
255 MR_Client c = it.next();
257 logger.info( "c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
258 MR_Client nc = new MR_Client( c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
259 nc.setFqtn(topic.getFqtn());
260 nc.setClientIdentity( c.getClientIdentity());
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()) {
268 topic.setClients(clients2);
271 Topic ntopic = checkForBridge( topic, err );
272 if ( ntopic == null ) {
273 topic.setStatus( DmaapObject_Status.INVALID);
274 if ( ! err.is2xx()) {
280 mr_topics.put( nFqtn, ntopic );
282 err.setCode(Status.OK.getStatusCode());
287 public Topic updateTopic( Topic topic, ApiError err ) {
288 logger.info( "Entry: updateTopic");
289 if ( topic.getFqtn().isEmpty()) {
292 Topic ntopic = checkForBridge( topic, err );
293 if ( ntopic == null ) {
294 topic.setStatus( DmaapObject_Status.INVALID);
295 if ( ! err.is2xx() ) {
300 mr_topics.put( ntopic.getFqtn(), ntopic );
302 err.setCode(Status.OK.getStatusCode());
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");
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();
319 clientService.removeMr_Client(c.getMrClientId(), false, apiError);
320 if ( ! apiError.is2xx()) {
324 apiError.setCode(Status.OK.getStatusCode());
325 return mr_topics.remove(pubId);
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" );
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 );
346 bridgeAdminTopic.setClients(clients);
348 TopicService ts = new TopicService();
349 ApiError err = new ApiError();
350 ts.addTopic(bridgeAdminTopic, err, true);
352 if ( err.is2xx() || err.getCode() == 409 ){
357 errorLogger.error( DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
362 public Topic checkForBridge( Topic topic, ApiError err ) {
364 if ( topic.getReplicationCase() == ReplicationType.REPLICATION_NONE ) {
365 topic.setStatus( DmaapObject_Status.VALID);
369 boolean anythingWrong = false;
371 Set<String> groups = clusters.getGroups();
372 for ( String g : groups ) {
373 anythingWrong |= buildBridge( topic, err, g );
375 if ( anythingWrong ) {
376 topic.setStatus( DmaapObject_Status.INVALID);
377 if ( ! err.is2xx() ) {
381 topic.setStatus( DmaapObject_Status.VALID);
386 private boolean buildBridge( Topic topic, ApiError err, String group ) {
388 boolean anythingWrong = false;
390 if ( group == null || group.isEmpty() ) {
391 graph = new Graph( topic.getClients(), true );
393 graph = new Graph( topic.getClients(), true, group );
395 MR_Cluster groupCentralCluster = null;
397 if ( graph.isEmpty() ) {
399 } else if ( group == null && topic.getReplicationCase().involvesFQDN() ) {
401 } else if ( ! graph.hasCentral() ) {
402 logger.warn( "Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
405 groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
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 );
416 String source = null;
417 String target = null;
420 * Provision Edge to Central bridges...
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;
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();
434 case REPLICATION_CENTRAL_TO_GLOBAL:
435 case REPLICATION_GLOBAL_TO_CENTRAL:
436 case REPLICATION_FQDN_TO_GLOBAL:
437 case REPLICATION_GLOBAL_TO_FQDN:
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();
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();
452 logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
453 anythingWrong = true;
455 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
456 + topic.getReplicationCase() );
457 err.setMessage("Unexpected value for ReplicationType");
461 } else if ( location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
463 * Provision Central to Global bridges
465 switch( topic.getReplicationCase() ) {
467 case REPLICATION_CENTRAL_TO_GLOBAL:
468 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
469 source = centralCname;
470 target = topic.getGlobalMrURL();
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;
478 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2F portion only
479 source = groupCentralCluster.getFqdn();
480 target = topic.getGlobalMrURL();
483 case REPLICATION_FQDN_TO_GLOBAL:
484 source = groupCentralCluster.getFqdn();
485 target = topic.getGlobalMrURL();
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();
494 case REPLICATION_FQDN_TO_EDGE:
495 case REPLICATION_EDGE_TO_FQDN:
496 case REPLICATION_EDGE_TO_CENTRAL:
497 case REPLICATION_CENTRAL_TO_EDGE:
500 logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
501 anythingWrong = true;
503 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
504 + topic.getReplicationCase() );
505 err.setMessage("Unexpected value for ReplicationType");
509 logger.warn( "dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
510 anythingWrong = true;
513 if ( source != null && target != null ) {
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 ) {
521 err.setFields( "mirror_maker.topic");
522 err.setMessage("Unexpected condition: " + ex );
523 anythingWrong = true;
530 return anythingWrong;
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.
541 public ReplicationType reviewTopic( Topic topic ) {
544 if ( topic.getNumClients() > 1 ) {
545 Graph graph = new Graph( topic.getClients(), false );
547 String centralFqdn = new String();
548 if ( graph.hasCentral() ) {
549 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
550 centralFqdn = p.getProperty("MR.CentralCname");
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 );
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;
570 return ReplicationType.REPLICATION_NONE;