8ade70f2a3caf6a96ba03f132f7647dd25333c0e
[dmaap/dbcapi.git] / src / main / java / org / onap / dmaap / dbcapi / service / TopicService.java
1 /*-
2  * ============LICENSE_START=======================================================
3  * org.onap.dmaap
4  * ================================================================================
5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
6  *
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
12  * 
13  *      http://www.apache.org/licenses/LICENSE-2.0
14  * 
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=========================================================
21  */
22
23 package org.onap.dmaap.dbcapi.service;
24
25 import java.util.ArrayList;
26 import java.util.Collection;
27 import java.util.Iterator;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.Set;
31
32 import javax.ws.rs.core.Response.Status;
33
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;
55
56 public class TopicService extends BaseLoggingClass {
57
58         
59
60         // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
61         private static String defaultGlobalMrHost;
62         
63         private Map<String, Topic> mr_topics = DatabaseClass.getTopics();
64         
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();
70         
71         private static String centralCname;
72
73
74         public TopicService(){
75                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
76                 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
77                 centralCname = p.getProperty("MR.CentralCname");
78
79                 
80                 logger.info( "TopicService properties: CentralCname=" + centralCname + 
81                                 "   defaultGlobarlMrHost=" + defaultGlobalMrHost  );
82         }
83         
84         public Map<String, Topic> getTopics() {                 
85                 return mr_topics;
86         }
87                 
88         public List<Topic> getAllTopics() {
89                 return getAllTopics( true );
90         }
91         public List<Topic> getAllTopicsWithoutClients() {
92                 return getAllTopics(false);
93         }
94         
95         private List<Topic> getAllTopics( Boolean withClients ) {
96                 ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
97                 if ( withClients ) {
98                         for( Topic topic: topics ) {
99                                 topic.setClients( clientService.getAllMrClients(topic.getFqtn()));
100                         }
101                 }
102                 return topics;
103         }
104         
105                 
106         public Topic getTopic( String key, ApiError apiError ) {        
107                 logger.info( "getTopic: key=" + key);
108                 Topic t = mr_topics.get( key );
109                 if ( t == null ) {
110                         apiError.setCode(Status.NOT_FOUND.getStatusCode());
111                         apiError.setFields( "fqtn");
112                         apiError.setMessage("topic with fqtn " + key + " not found");
113                         return null;
114                 }
115                 t.setClients( clientService.getAllMrClients( key ));
116                 apiError.setCode(Status.OK.getStatusCode());
117                 return t;
118         }
119         
120         private void aafTopicSetup(Topic topic, ApiError err ) {
121
122                 String t = dmaapSvc.getTopicPerm();
123                 if ( t == null ) {
124                         err.setCode(500);
125                         err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)"  );
126                         err.setFields("topicNsRoot");
127                         return;
128                 }
129
130                 // establish AAF Connection using TopicMgr identity
131                 AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
132                 
133
134                 
135                 // create AAF namespace for this topic
136                 AafNamespace ns = new AafNamespace( topic.getFqtn(), aaf.getIdentity());
137                 {
138                         int rc = aaf.addNamespace( ns );
139                         if ( rc != 201 && rc != 409 ) {
140                                 err.setCode(500);
141                                 err.setMessage("Unexpected response from AAF:" + rc );
142                                 err.setFields("namespace:" + topic.getFqtn() + " identity="+ aaf.getIdentity());
143                                 return;
144                         }
145                 }
146                 
147                 // create AAF Roles for MR clients of this topic
148                 String rn = "publisher";
149                 AafRole pubRole = new AafRole( topic.getFqtn(), rn );
150                 int rc = aaf.addRole( pubRole );
151                 if ( rc != 201 && rc != 409 ) {
152                         err.setCode(500);
153                         err.setMessage("Unexpected response from AAF:" + rc );
154                         err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
155                         return;
156                 }
157                 topic.setPublisherRole( pubRole.getFullyQualifiedRole() );
158                 
159                 rn = "subscriber";
160                 AafRole subRole = new AafRole( topic.getFqtn(), rn );
161                 rc = aaf.addRole( subRole );
162                 if ( rc != 201 && rc != 409 ) {
163                         err.setCode(500);
164                         err.setMessage("Unexpected response from AAF:" + rc );
165                         err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
166                         return;
167                 }
168                 topic.setSubscriberRole( subRole.getFullyQualifiedRole() );
169         
170                 
171                 // create AAF perms checked by MR
172                 String instance = ":topic." + topic.getFqtn();
173                 String[] actions = { "pub", "sub", "view" };
174                 for ( String action : actions ){
175                         DmaapPerm perm = new DmaapPerm( t, instance, action );
176                         rc = aaf.addPerm( perm );
177                         if ( rc != 201 && rc != 409 ) {
178                                 err.setCode(500);
179                                 err.setMessage("Unexpected response from AAF:" + rc );
180                                 err.setFields("t="+t + " instance="+ instance + " action="+ action);
181                                 return;
182                         }
183                         // Grant perms to our default Roles
184                         if ( action.equals( "pub") || action.equals( "view") ) {
185                                 DmaapGrant g = new DmaapGrant( perm, pubRole.getFullyQualifiedRole() );
186                                 rc = aaf.addGrant( g );
187                                 if ( rc != 201 && rc != 409 ) {
188                                         err.setCode(rc);
189                                         err.setMessage( "Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole() );
190                                         logger.warn( err.getMessage());
191                                         return;
192                                 } 
193                         }
194                         if ( action.equals( "sub") || action.equals( "view") ) {
195                                 DmaapGrant g = new DmaapGrant( perm, subRole.getFullyQualifiedRole() );
196                                 rc = aaf.addGrant( g );
197                                 if ( rc != 201 && rc != 409 ) {
198                                         err.setCode(rc);
199                                         err.setMessage( "Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole() );
200                                         logger.warn( err.getMessage());
201                                         return;
202                                 } 
203                         }
204
205                 }
206         }
207
208         public Topic addTopic( Topic topic, ApiError err, Boolean useExisting ) {
209                 logger.info( "Entry: addTopic");
210                 logger.info( "Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle() );
211                 String nFqtn =  topic.genFqtn();
212                 logger.info( "FQTN=" + nFqtn );
213                 Topic pTopic = getTopic( nFqtn, err );
214                 if ( pTopic != null ) {
215                         String t = "topic already exists: " + nFqtn;
216                         logger.info( t );
217                         if (  useExisting ) {
218                                 err.setCode(Status.OK.getStatusCode());
219                                 return pTopic;
220                         }
221                         err.setMessage( t );
222                         err.setFields( "fqtn");
223                         err.setCode(Status.CONFLICT.getStatusCode());
224                         return null;
225                 }
226                 err.reset();  // err filled with NOT_FOUND is expected case, but don't want to litter...
227
228                 topic.setFqtn( nFqtn );
229                 
230                 aafTopicSetup( topic, err );
231                 if ( err.getCode() >= 400 ) {
232                         return null;
233                 }       
234
235                 if ( topic.getReplicationCase().involvesGlobal() ) {
236                         if ( topic.getGlobalMrURL() == null ) {
237                                 topic.setGlobalMrURL(defaultGlobalMrHost);
238                         }
239                         if ( ! Fqdn.isValid( topic.getGlobalMrURL())) {
240                                 logger.error( "GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
241                                 topic.setStatus( DmaapObject_Status.INVALID);
242                                 err.setCode(500);
243                                 err.setMessage("Value is not a valid FQDN:" +  topic.getGlobalMrURL() );
244                                 err.setFields("globalMrURL");
245         
246                                 return null;
247                         }
248                 }
249
250
251                 if ( topic.getNumClients() > 0 ) {
252                         ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
253                 
254         
255                         ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
256                         for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
257                                 MR_Client c = it.next();
258
259                                 logger.info( "c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
260                                 MR_Client nc = new MR_Client( c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
261                                 nc.setFqtn(topic.getFqtn());
262                                 nc.setClientIdentity( c.getClientIdentity());
263                                 logger.info( "nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
264                                 clients2.add( clientService.addMr_Client(nc, topic, err));
265                                 if ( ! err.is2xx()) {
266                                         return null;
267                                 }
268                         }
269
270                         topic.setClients(clients2);
271                 }
272
273                 Topic ntopic = checkForBridge( topic, err );
274                 if ( ntopic == null ) {
275                         topic.setStatus( DmaapObject_Status.INVALID);
276                         if ( ! err.is2xx()) {
277                                 return null;
278                         }
279                 }
280
281                 
282                 mr_topics.put( nFqtn, ntopic );
283
284                 err.setCode(Status.OK.getStatusCode());
285                 return ntopic;
286         }
287         
288                 
289         public Topic updateTopic( Topic topic, ApiError err ) {
290                 logger.info( "Entry: updateTopic");
291                 if ( topic.getFqtn().isEmpty()) {
292                         return null;
293                 }
294                 Topic ntopic = checkForBridge( topic, err );
295                 if ( ntopic == null ) {
296                         topic.setStatus( DmaapObject_Status.INVALID);
297                         if ( ! err.is2xx() ) {
298                                 return null;
299                         }
300                 }
301                 if(ntopic != null) {
302                         mr_topics.put( ntopic.getFqtn(), ntopic );
303                 }
304                 err.setCode(Status.OK.getStatusCode());
305                 return ntopic;
306         }
307                 
308         public Topic removeTopic( String pubId, ApiError apiError ) {
309                 Topic topic = mr_topics.get(pubId);
310                 if ( topic == null ) {
311                         apiError.setCode(Status.NOT_FOUND.getStatusCode());
312                         apiError.setMessage("Topic " + pubId + " does not exist");
313                         apiError.setFields("fqtn");
314                         return null;
315                 }
316                 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients( pubId ));
317                 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
318                         MR_Client c = it.next();
319                         
320         
321                         clientService.removeMr_Client(c.getMrClientId(), false, apiError);
322                         if ( ! apiError.is2xx()) {
323                                 return null;
324                         }
325                 }
326                 apiError.setCode(Status.OK.getStatusCode());
327                 return mr_topics.remove(pubId);
328         }       
329         public static ApiError setBridgeClientPerms( MR_Cluster node ) {
330                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
331                 String mmProvRole = p.getProperty("MM.ProvRole");
332                 String mmAgentRole = p.getProperty("MM.AgentRole");
333                 String[] Roles = { mmProvRole, mmAgentRole };
334                 String[] actions = { "view", "pub", "sub" };
335                 Topic bridgeAdminTopic = new Topic().init();
336                 bridgeAdminTopic.setTopicName( dmaapSvc.getBridgeAdminFqtn() );
337                 bridgeAdminTopic.setTopicDescription( "RESERVED topic for MirroMaker Provisioning");
338                 bridgeAdminTopic.setOwner( "DBC" );
339                 
340                 ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
341                 for( String role: Roles ) {
342                         MR_Client client = new MR_Client();
343                         client.setAction(actions);
344                         client.setClientRole(role);
345                         client.setDcaeLocationName( node.getDcaeLocationName());
346                         clients.add( client );
347                 }
348                 bridgeAdminTopic.setClients(clients);
349                 
350                 TopicService ts = new TopicService();
351                 ApiError err = new ApiError();
352                 ts.addTopic(bridgeAdminTopic, err, true);
353                 
354                 if ( err.is2xx() || err.getCode() == 409 ){
355                         err.setCode(200);
356                         return err;
357                 }
358                 
359                 errorLogger.error( DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR,  bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
360                 return err;
361         }       
362         
363         
364         public Topic checkForBridge( Topic topic, ApiError err ) {
365                 
366                 if ( topic.getReplicationCase() == ReplicationType.REPLICATION_NONE ) {
367                         topic.setStatus( DmaapObject_Status.VALID);
368                         return topic;   
369                 }
370                 
371                 boolean anythingWrong = false;
372                 
373                 Set<String> groups = clusters.getGroups();
374                 for ( String g : groups ) {
375                         anythingWrong |= buildBridge( topic, err, g );
376                 }
377                 if ( anythingWrong ) {
378                         topic.setStatus( DmaapObject_Status.INVALID);
379                         if ( ! err.is2xx() ) {
380                                 return null;
381                         }       
382                 } else {
383                         topic.setStatus( DmaapObject_Status.VALID);
384                 }
385                 return topic;
386         }
387                 
388         private boolean buildBridge( Topic topic, ApiError err, String group ) {
389
390                 boolean anythingWrong = false;
391                 Graph graph;
392                 if ( group == null || group.isEmpty() ) {
393                         graph = new Graph( topic.getClients(), true );
394                 } else {
395                         graph = new Graph( topic.getClients(), true, group );
396                 }
397                 MR_Cluster groupCentralCluster = null;
398                 
399                 if ( graph.isEmpty() ) {
400                         return false;
401                 } else if ( group == null &&  topic.getReplicationCase().involvesFQDN() ) {
402                         return false;
403                 } else if ( ! graph.hasCentral() ) {
404                         logger.warn( "Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
405                         return true;
406                 } else {
407                         groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
408                 }
409                 Collection<String> clientLocations = graph.getKeys();
410                 for( String loc : clientLocations ) {
411                         logger.info( "loc=" + loc );
412                         DcaeLocation location = locations.getDcaeLocation(loc);
413                         MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
414                         logger.info( "cluster=" + cluster );
415
416                         
417                                 
418                         String source = null;
419                         String target = null;
420                         
421                         /*
422                          * Provision Edge to Central bridges...
423                          */
424                         if ( ! location.isCentral()  && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
425                                 switch( topic.getReplicationCase() ) {
426                                 case REPLICATION_EDGE_TO_CENTRAL:
427                                 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:  // NOTE: this is for E2C portion only
428                                         source = cluster.getFqdn();
429                                         target = centralCname;
430                                         break;
431                                 case REPLICATION_CENTRAL_TO_EDGE:
432                                 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for C2E portion only
433                                         source = centralCname;
434                                         target = cluster.getFqdn();
435                                         break;
436                                 case REPLICATION_CENTRAL_TO_GLOBAL:
437                                 case REPLICATION_GLOBAL_TO_CENTRAL:
438                                 case REPLICATION_FQDN_TO_GLOBAL:
439                                 case REPLICATION_GLOBAL_TO_FQDN:
440                                         break;
441
442                                 case REPLICATION_EDGE_TO_FQDN:
443                                 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2C portion only
444                                         source = cluster.getFqdn();
445                                         target = groupCentralCluster.getFqdn();
446                                         break;
447                                 case REPLICATION_FQDN_TO_EDGE:
448                                 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for F2E portion only
449                                         source = groupCentralCluster.getFqdn();
450                                         target = cluster.getFqdn();
451                                         break;
452
453                                 default:
454                                         logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
455                                         anythingWrong = true;
456                                         err.setCode(400);
457                                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
458                                                         + topic.getReplicationCase() );
459                                         err.setMessage("Unexpected value for ReplicationType");
460                                         continue;
461                                 }
462
463                         } else if ( location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
464                                 /*
465                                  * Provision Central to Global bridges
466                                  */
467                                 switch( topic.getReplicationCase() ) {
468
469                                 case REPLICATION_CENTRAL_TO_GLOBAL:
470                                 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
471                                         source = centralCname;
472                                         target = topic.getGlobalMrURL();
473                                         break;
474                                 case REPLICATION_GLOBAL_TO_CENTRAL:
475                                 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for G2C portion only
476                                         source = topic.getGlobalMrURL();
477                                         target = centralCname;
478                                         break;
479
480                                 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
481                                         source = groupCentralCluster.getFqdn();
482                                         target = topic.getGlobalMrURL();
483                                         break;
484
485                                 case REPLICATION_FQDN_TO_GLOBAL:
486                                         source = groupCentralCluster.getFqdn();
487                                         target = topic.getGlobalMrURL();
488                                         break;
489                                         
490                                 case REPLICATION_GLOBAL_TO_FQDN:
491                                 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for G2F portion only
492                                         source = topic.getGlobalMrURL();
493                                         target = groupCentralCluster.getFqdn();
494                                         break;
495
496                                 case REPLICATION_FQDN_TO_EDGE:
497                                 case REPLICATION_EDGE_TO_FQDN:
498                                 case REPLICATION_EDGE_TO_CENTRAL:
499                                 case REPLICATION_CENTRAL_TO_EDGE:
500                                         break;
501                                 default:
502                                         logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
503                                         anythingWrong = true;
504                                         err.setCode(400);
505                                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
506                                                         + topic.getReplicationCase() );
507                                         err.setMessage("Unexpected value for ReplicationType");
508                                         continue;
509                                 }                               
510                         } else {
511                                 logger.warn( "dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
512                                 anythingWrong = true;
513                                 continue;
514                         }
515                         if ( source != null && target != null ) {
516                                 try { 
517                                         logger.info( "Create a MM from " + source + " to " + target );
518                                         MirrorMaker mm = bridge.getNextMM( source, target, topic.getFqtn());
519                                         mm.addTopic(topic.getFqtn());
520                                         bridge.updateMirrorMaker(mm);
521                                 } catch ( Exception ex ) {
522                                         err.setCode(500);
523                                         err.setFields( "mirror_maker.topic");
524                                         err.setMessage("Unexpected condition: " + ex );
525                                         anythingWrong = true;
526                                         break;
527                                 }
528                         }                       
529
530                         
531                 }
532                 return  anythingWrong;
533
534         }
535         
536         
537         /*
538          * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
539          * This was determined automatically based on presence of edge publishers and central subscribers.
540          * The following method is a modification of that original logic, to preserve some backwards compatibility, 
541          * i.e. to be used when no ReplicationType is specified.
542          */
543         public ReplicationType reviewTopic( Topic topic ) {
544         
545                 
546                 if ( topic.getNumClients() > 1 ) {
547                         Graph graph = new Graph( topic.getClients(), false );
548                         
549                         String centralFqdn = new String();
550                         if ( graph.hasCentral() ) {
551                                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
552                                 centralFqdn = p.getProperty("MR.CentralCname");
553                         }
554
555                         Collection<String> locations = graph.getKeys();
556                         for( String loc : locations ) {
557                                 logger.info( "loc=" + loc );
558                                 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
559                                 if ( cluster == null ) {
560                                         logger.info( "No MR cluster for location " + loc );
561                                         continue;
562                                 }
563                                 if ( graph.hasCentral() &&  ! graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
564                                         logger.info( "Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn );
565                                         return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
566                                         
567                                 }
568                                 
569                         }
570                 }
571         
572                 return ReplicationType.REPLICATION_NONE;
573         }
574
575 }