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.AafService;
33 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
34 import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
35 import org.onap.dmaap.dbcapi.database.DatabaseClass;
36 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
37 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
38 import org.onap.dmaap.dbcapi.model.ApiError;
39 import org.onap.dmaap.dbcapi.model.DcaeLocation;
40 import org.onap.dmaap.dbcapi.model.Dmaap;
41 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
42 import org.onap.dmaap.dbcapi.model.MR_Client;
43 import org.onap.dmaap.dbcapi.model.MR_Cluster;
44 import org.onap.dmaap.dbcapi.model.MirrorMaker;
45 import org.onap.dmaap.dbcapi.model.ReplicationType;
46 import org.onap.dmaap.dbcapi.model.Topic;
47 import org.onap.dmaap.dbcapi.util.DmaapConfig;
48 import org.onap.dmaap.dbcapi.util.Fqdn;
49 import org.onap.dmaap.dbcapi.util.Graph;
51 public class TopicService extends BaseLoggingClass {
55 // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
56 private static String defaultGlobalMrHost;
58 private Map<String, Topic> mr_topics = DatabaseClass.getTopics();
60 private static DmaapService dmaapSvc = new DmaapService();
61 private static Dmaap dmaap = new DmaapService().getDmaap();
62 private MR_ClientService clientService = new MR_ClientService();
63 private MR_ClusterService clusters = new MR_ClusterService();
64 private DcaeLocationService locations = new DcaeLocationService();
65 private MirrorMakerService bridge = new MirrorMakerService();
67 private static String centralCname;
69 public TopicService(){
70 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
71 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
72 centralCname = p.getProperty("MR.CentralCname");
74 logger.info( "TopicService properties: CentralCname=" + centralCname +
75 " defaultGlobarlMrHost=" + defaultGlobalMrHost );
78 public Map<String, Topic> getTopics() {
82 public List<Topic> getAllTopics() {
83 return getAllTopics( true );
85 public List<Topic> getAllTopicsWithoutClients() {
86 return getAllTopics(false);
89 private List<Topic> getAllTopics( Boolean withClients ) {
90 ArrayList<Topic> topics = new ArrayList<Topic>(mr_topics.values());
92 for( Topic topic: topics ) {
93 topic.setClients( clientService.getAllMrClients(topic.getFqtn()));
100 public Topic getTopic( String key, ApiError apiError ) {
101 logger.info( "getTopic: key=" + key);
102 Topic t = mr_topics.get( key );
104 apiError.setCode(Status.NOT_FOUND.getStatusCode());
105 apiError.setFields( "fqtn");
106 apiError.setMessage("topic with fqtn " + key + " not found");
109 t.setClients( clientService.getAllMrClients( key ));
110 apiError.setCode(Status.OK.getStatusCode());
114 public Topic addTopic( Topic topic, ApiError err, Boolean useExisting ) {
115 logger.info( "Entry: addTopic");
116 logger.info( "Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle() );
117 String nFqtn = topic.genFqtn();
118 logger.info( "FQTN=" + nFqtn );
119 Topic pTopic = getTopic( nFqtn, err );
120 if ( pTopic != null ) {
121 String t = "topic already exists: " + nFqtn;
124 err.setCode(Status.OK.getStatusCode());
128 err.setFields( "fqtn");
129 err.setCode(Status.CONFLICT.getStatusCode());
132 err.reset(); // err filled with NOT_FOUND is expected case, but don't want to litter...
134 topic.setFqtn( nFqtn );
136 AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
138 String t = dmaapSvc.getTopicPerm();
140 String instance = ":topic." + topic.getFqtn();
142 String[] actions = { "pub", "sub", "view" };
143 for ( String action : actions ){
144 DmaapPerm perm = new DmaapPerm( t, instance, action );
145 int rc = aaf.addPerm( perm );
146 if ( rc != 201 && rc != 409 ) {
148 err.setMessage("Unexpected response from AAF:" + rc );
149 err.setFields("t="+t + " instance="+ instance + " action="+ action);
153 if ( topic.getReplicationCase().involvesGlobal() ) {
154 if ( topic.getGlobalMrURL() == null ) {
155 topic.setGlobalMrURL(defaultGlobalMrHost);
157 if ( ! Fqdn.isValid( topic.getGlobalMrURL())) {
158 logger.error( "GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
159 topic.setStatus( DmaapObject_Status.INVALID);
161 err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL() );
162 err.setFields("globalMrURL");
169 if ( topic.getNumClients() > 0 ) {
170 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
173 ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
174 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
175 MR_Client c = it.next();
177 logger.info( "c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
178 MR_Client nc = new MR_Client( c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
179 nc.setFqtn(topic.getFqtn());
180 logger.info( "nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
181 clients2.add( clientService.addMr_Client(nc, topic, err));
182 if ( ! err.is2xx()) {
187 topic.setClients(clients2);
190 Topic ntopic = checkForBridge( topic, err );
191 if ( ntopic == null ) {
192 topic.setStatus( DmaapObject_Status.INVALID);
193 if ( ! err.is2xx()) {
199 mr_topics.put( nFqtn, ntopic );
201 err.setCode(Status.OK.getStatusCode());
206 public Topic updateTopic( Topic topic, ApiError err ) {
207 logger.info( "Entry: updateTopic");
208 if ( topic.getFqtn().isEmpty()) {
211 Topic ntopic = checkForBridge( topic, err );
212 if ( ntopic == null ) {
213 topic.setStatus( DmaapObject_Status.INVALID);
214 if ( ! err.is2xx() ) {
219 mr_topics.put( ntopic.getFqtn(), ntopic );
221 err.setCode(Status.OK.getStatusCode());
225 public Topic removeTopic( String pubId, ApiError apiError ) {
226 Topic topic = mr_topics.get(pubId);
227 if ( topic == null ) {
228 apiError.setCode(Status.NOT_FOUND.getStatusCode());
229 apiError.setMessage("Topic " + pubId + " does not exist");
230 apiError.setFields("fqtn");
233 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients( pubId ));
234 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
235 MR_Client c = it.next();
238 clientService.removeMr_Client(c.getMrClientId(), false, apiError);
239 if ( ! apiError.is2xx()) {
243 apiError.setCode(Status.OK.getStatusCode());
244 return mr_topics.remove(pubId);
246 public static ApiError setBridgeClientPerms( MR_Cluster node ) {
247 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
248 String mmProvRole = p.getProperty("MM.ProvRole");
249 String mmAgentRole = p.getProperty("MM.AgentRole");
250 String[] Roles = { mmProvRole, mmAgentRole };
251 String[] actions = { "view", "pub", "sub" };
252 Topic bridgeAdminTopic = new Topic();
253 bridgeAdminTopic.setTopicName( dmaapSvc.getBridgeAdminFqtn() );
254 bridgeAdminTopic.setTopicDescription( "RESERVED topic for MirroMaker Provisioning");
255 bridgeAdminTopic.setOwner( "DBC" );
256 ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
257 for( String role: Roles ) {
258 MR_Client client = new MR_Client();
259 client.setAction(actions);
260 client.setClientRole(role);
261 client.setDcaeLocationName( node.getDcaeLocationName());
262 clients.add( client );
264 bridgeAdminTopic.setClients(clients);
266 TopicService ts = new TopicService();
267 ApiError err = new ApiError();
268 ts.addTopic(bridgeAdminTopic, err, true);
270 if ( err.is2xx() || err.getCode() == 409 ){
275 errorLogger.error( DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
280 public Topic checkForBridge( Topic topic, ApiError err ) {
282 if ( topic.getReplicationCase() == ReplicationType.REPLICATION_NONE ) {
283 topic.setStatus( DmaapObject_Status.VALID);
287 boolean anythingWrong = false;
289 Set<String> groups = clusters.getGroups();
290 for ( String g : groups ) {
291 anythingWrong |= buildBridge( topic, err, g );
293 if ( anythingWrong ) {
294 topic.setStatus( DmaapObject_Status.INVALID);
295 if ( ! err.is2xx() ) {
299 topic.setStatus( DmaapObject_Status.VALID);
304 private boolean buildBridge( Topic topic, ApiError err, String group ) {
306 boolean anythingWrong = false;
308 if ( group == null || group.isEmpty() ) {
309 graph = new Graph( topic.getClients(), true );
311 graph = new Graph( topic.getClients(), true, group );
313 MR_Cluster groupCentralCluster = null;
315 if ( graph.isEmpty() ) {
317 } else if ( group == null && topic.getReplicationCase().involvesFQDN() ) {
319 } else if ( ! graph.hasCentral() ) {
320 logger.warn( "Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
323 groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
325 Collection<String> clientLocations = graph.getKeys();
326 for( String loc : clientLocations ) {
327 logger.info( "loc=" + loc );
328 DcaeLocation location = locations.getDcaeLocation(loc);
329 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
330 logger.info( "cluster=" + cluster );
334 String source = null;
335 String target = null;
338 * Provision Edge to Central bridges...
340 if ( ! location.isCentral() && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
341 switch( topic.getReplicationCase() ) {
342 case REPLICATION_EDGE_TO_CENTRAL:
343 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL: // NOTE: this is for E2C portion only
344 source = cluster.getFqdn();
345 target = centralCname;
347 case REPLICATION_CENTRAL_TO_EDGE:
348 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for C2E portion only
349 source = centralCname;
350 target = cluster.getFqdn();
352 case REPLICATION_CENTRAL_TO_GLOBAL:
353 case REPLICATION_GLOBAL_TO_CENTRAL:
354 case REPLICATION_FQDN_TO_GLOBAL:
355 case REPLICATION_GLOBAL_TO_FQDN:
358 case REPLICATION_EDGE_TO_FQDN:
359 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2C portion only
360 source = cluster.getFqdn();
361 target = groupCentralCluster.getFqdn();
363 case REPLICATION_FQDN_TO_EDGE:
364 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for F2E portion only
365 source = groupCentralCluster.getFqdn();
366 target = cluster.getFqdn();
370 logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
371 anythingWrong = true;
373 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
374 + topic.getReplicationCase() );
375 err.setMessage("Unexpected value for ReplicationType");
379 } else if ( location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
381 * Provision Central to Global bridges
383 switch( topic.getReplicationCase() ) {
385 case REPLICATION_CENTRAL_TO_GLOBAL:
386 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
387 source = centralCname;
388 target = topic.getGlobalMrURL();
390 case REPLICATION_GLOBAL_TO_CENTRAL:
391 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE: // NOTE: this is for G2C portion only
392 source = topic.getGlobalMrURL();
393 target = centralCname;
396 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL: // NOTE: this is for E2F portion only
397 source = groupCentralCluster.getFqdn();
398 target = topic.getGlobalMrURL();
401 case REPLICATION_FQDN_TO_GLOBAL:
402 source = groupCentralCluster.getFqdn();
403 target = topic.getGlobalMrURL();
406 case REPLICATION_GLOBAL_TO_FQDN:
407 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE: // NOTE: this is for G2F portion only
408 source = topic.getGlobalMrURL();
409 target = groupCentralCluster.getFqdn();
412 case REPLICATION_FQDN_TO_EDGE:
413 case REPLICATION_EDGE_TO_FQDN:
414 case REPLICATION_EDGE_TO_CENTRAL:
415 case REPLICATION_CENTRAL_TO_EDGE:
418 logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
419 anythingWrong = true;
421 err.setFields("topic=" + topic.genFqtn() + " replicationCase="
422 + topic.getReplicationCase() );
423 err.setMessage("Unexpected value for ReplicationType");
427 logger.warn( "dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
428 anythingWrong = true;
431 if ( source != null && target != null ) {
433 logger.info( "Create a MM from " + source + " to " + target );
434 MirrorMaker mm = bridge.getMirrorMaker( source, target);
436 mm = new MirrorMaker(source, target);
438 mm.addTopic(topic.getFqtn());
439 bridge.updateMirrorMaker(mm);
440 } catch ( Exception ex ) {
442 err.setFields( "mirror_maker.topic");
443 err.setMessage("Unexpected condition: " + ex );
444 anythingWrong = true;
451 return anythingWrong;
456 * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
457 * This was determined automatically based on presence of edge publishers and central subscribers.
458 * The following method is a modification of that original logic, to preserve some backwards compatibility,
459 * i.e. to be used when no ReplicationType is specified.
461 public ReplicationType reviewTopic( Topic topic ) {
464 if ( topic.getNumClients() > 1 ) {
465 Graph graph = new Graph( topic.getClients(), false );
467 String centralFqdn = new String();
468 if ( graph.hasCentral() ) {
469 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
470 centralFqdn = p.getProperty("MR.CentralCname");
473 Collection<String> locations = graph.getKeys();
474 for( String loc : locations ) {
475 logger.info( "loc=" + loc );
476 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
477 if ( cluster == null ) {
478 logger.info( "No MR cluster for location " + loc );
481 if ( graph.hasCentral() && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
482 logger.info( "Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn );
483 return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
490 return ReplicationType.REPLICATION_NONE;