gerrit.onap Code Review - ccsdk/oran.git/blob

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * ONAP : ccsdk oran
   4  * ======================================================================
   5  * Copyright (C) 2019-2022 Nordix Foundation. 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  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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===================================
  19  */
  20
  21 package org.onap.ccsdk.oran.a1policymanagementservice.repository;
  22
  23 import com.google.gson.Gson;
  24 import com.google.gson.GsonBuilder;
  25
  26 import java.lang.invoke.MethodHandles;
  27 import java.time.Instant;
  28 import java.util.ArrayList;
  29 import java.util.Collection;
  30 import java.util.HashMap;
  31 import java.util.List;
  32 import java.util.Map;
  33 import java.util.Objects;
  34 import java.util.Set;
  35 import java.util.Vector;
  36
  37 import lombok.Builder;
  38 import lombok.Getter;
  39
  40 import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationConfig;
  41 import org.onap.ccsdk.oran.a1policymanagementservice.datastore.DataStore;
  42 import org.onap.ccsdk.oran.a1policymanagementservice.exceptions.EntityNotFoundException;
  43 import org.slf4j.Logger;
  44 import org.slf4j.LoggerFactory;
  45 import org.springframework.beans.factory.annotation.Autowired;
  46
  47 import reactor.core.publisher.Flux;
  48 import reactor.core.publisher.Mono;
  49 import reactor.util.annotation.Nullable;
  50
  51 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
  52 public class Policies {
  53
  54     @Getter
  55     @Builder
  56     private static class PersistentPolicyInfo {
  57         private String id;
  58         private String json;
  59         private String ownerServiceId;
  60         private String ricId;
  61         private String typeId;
  62         private String statusNotificationUri;
  63         private boolean isTransient;
  64         private String lastModified;
  65     }
  66
  67     private final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  68     private Map<String, Policy> policiesId = new HashMap<>();
  69     private MultiMap<Policy> policiesRic = new MultiMap<>();
  70     private MultiMap<Policy> policiesService = new MultiMap<>();
  71     private MultiMap<Policy> policiesType = new MultiMap<>();
  72     private final DataStore dataStore;
  73
  74     private final ApplicationConfig appConfig;
  75     private static Gson gson = new GsonBuilder().create();
  76
  77     public Policies(@Autowired ApplicationConfig appConfig) {
  78         this.appConfig = appConfig;
  79         this.dataStore = DataStore.create(appConfig, "policies");
  80     }
  81
  82     public Flux<Policy> restoreFromDatabase(Ric ric, PolicyTypes types) {
  83         return dataStore.createDataStore() //
  84                 .flatMapMany(x -> dataStore.listObjects(getPath(ric))) //
  85                 .flatMap(dataStore::readObject) //
  86                 .map(String::new) //
  87                 .map(json -> gson.fromJson(json, PersistentPolicyInfo.class)) //
  88                 .map(policyInfo -> toPolicy(policyInfo, ric, types)) //
  89                 .doOnNext(this::put) //
  90                 .filter(Objects::nonNull) //
  91                 .doOnError(t -> logger.warn("Could not restore policy database for RIC: {}, reason : {}", ric.id(),
  92                         t.getMessage())) //
  93                 .doFinally(sig -> logger.debug("Restored policy database for RIC: {}, number of policies: {}", ric.id(),
  94                         this.policiesRic.get(ric.id()).size())) //
  95                 .onErrorResume(t -> Flux.empty()) //
  96         ;
  97     }
  98
  99     public synchronized void put(Policy policy) {
 100         policiesId.put(policy.getId(), policy);
 101         policiesRic.put(policy.getRic().id(), policy.getId(), policy);
 102         policiesService.put(policy.getOwnerServiceId(), policy.getId(), policy);
 103         policiesType.put(policy.getType().getId(), policy.getId(), policy);
 104         if (!policy.isTransient()) {
 105             store(policy);
 106         }
 107     }
 108
 109     public synchronized boolean containsPolicy(String id) {
 110         return policiesId.containsKey(id);
 111     }
 112
 113     public synchronized Policy get(String id) {
 114         return policiesId.get(id);
 115     }
 116
 117     public synchronized Policy getPolicy(String id) throws EntityNotFoundException {
 118         Policy p = policiesId.get(id);
 119         if (p == null) {
 120             throw new EntityNotFoundException("Could not find policy: " + id);
 121         }
 122         return p;
 123     }
 124
 125     public synchronized Collection<Policy> getAll() {
 126         return new Vector<>(policiesId.values());
 127     }
 128
 129     public synchronized Collection<Policy> getForService(String service) {
 130         return policiesService.get(service);
 131     }
 132
 133     public synchronized Collection<Policy> getForRic(String ric) {
 134         return policiesRic.get(ric);
 135     }
 136
 137     public synchronized Collection<Policy> getForType(String type) {
 138         return policiesType.get(type);
 139     }
 140
 141     public synchronized Policy removeId(String id) {
 142         Policy p = policiesId.get(id);
 143         if (p != null) {
 144             remove(p);
 145         }
 146         return p;
 147     }
 148
 149     public synchronized void remove(Policy policy) {
 150         if (!policy.isTransient()) {
 151             dataStore.deleteObject(getPath(policy)).subscribe();
 152         }
 153         policiesId.remove(policy.getId());
 154         policiesRic.remove(policy.getRic().id(), policy.getId());
 155         policiesService.remove(policy.getOwnerServiceId(), policy.getId());
 156         policiesType.remove(policy.getType().getId(), policy.getId());
 157     }
 158
 159     public synchronized void removePoliciesForRic(String ricId) {
 160         Collection<Policy> policiesForRic = getForRic(ricId);
 161         for (Policy policy : policiesForRic) {
 162             remove(policy);
 163         }
 164     }
 165
 166     public Collection<Policy> filterPolicies(@Nullable String typeId, @Nullable String ricId,
 167             @Nullable String serviceId, @Nullable String typeName) {
 168
 169         if (typeId != null) {
 170             return filter(this.getForType(typeId), null, ricId, serviceId, typeName);
 171         } else if (serviceId != null) {
 172             return filter(this.getForService(serviceId), typeId, ricId, null, typeName);
 173         } else if (ricId != null) {
 174             return filter(this.getForRic(ricId), typeId, null, serviceId, typeName);
 175         } else {
 176             return filter(this.getAll(), typeId, ricId, serviceId, typeName);
 177         }
 178     }
 179
 180     public synchronized int size() {
 181         return policiesId.size();
 182     }
 183
 184     public synchronized void clear() {
 185         while (policiesId.size() > 0) {
 186             Set<String> keys = policiesId.keySet();
 187             removeId(keys.iterator().next());
 188         }
 189         dataStore.deleteAllObjects().onErrorResume(t -> Mono.empty()).subscribe();
 190     }
 191
 192     public void store(Policy policy) {
 193
 194         byte[] bytes = gson.toJson(toStorageObject(policy)).getBytes();
 195         this.dataStore.writeObject(this.getPath(policy), bytes) //
 196                 .doOnError(t -> logger.error("Could not store job in S3, reason: {}", t.getMessage())) //
 197                 .subscribe();
 198     }
 199
 200     private boolean isMatch(String filterValue, String actualValue) {
 201         return filterValue == null || actualValue.equals(filterValue);
 202     }
 203
 204     private boolean isTypeMatch(Policy policy, @Nullable String typeName) {
 205         return (typeName == null) || policy.getType().getTypeId().getName().equals(typeName);
 206     }
 207
 208     private Collection<Policy> filter(Collection<Policy> collection, String typeId, String ricId, String serviceId,
 209             String typeName) {
 210         if (typeId == null && ricId == null && serviceId == null && typeName == null) {
 211             return collection;
 212         }
 213         List<Policy> filtered = new ArrayList<>(collection.size());
 214         for (Policy p : collection) {
 215             if (isMatch(typeId, p.getType().getId()) && isMatch(ricId, p.getRic().id())
 216                     && isMatch(serviceId, p.getOwnerServiceId()) && isTypeMatch(p, typeName)) {
 217                 filtered.add(p);
 218             }
 219         }
 220         return filtered;
 221     }
 222
 223     private PersistentPolicyInfo toStorageObject(Policy p) {
 224         return PersistentPolicyInfo.builder() //
 225                 .id(p.getId()) //
 226                 .json(p.getJson()) //
 227                 .ownerServiceId(p.getOwnerServiceId()) //
 228                 .ricId(p.getRic().id()) //
 229                 .statusNotificationUri(p.getStatusNotificationUri()) //
 230                 .typeId(p.getType().getId()) //
 231                 .isTransient(p.isTransient()) //
 232                 .lastModified(p.getLastModified().toString()) //
 233                 .build();
 234     }
 235
 236     private Policy toPolicy(PersistentPolicyInfo p, Ric ric, PolicyTypes types) {
 237         try {
 238             return Policy.builder() //
 239                     .id(p.getId()) //
 240                     .isTransient(p.isTransient()) //
 241                     .json(p.getJson()) //
 242                     .lastModified(Instant.parse(p.lastModified)) //
 243                     .ownerServiceId(p.getOwnerServiceId()) //
 244                     .ric(ric) //
 245                     .statusNotificationUri(p.getStatusNotificationUri()) //
 246                     .type(types.getType(p.getTypeId())) //
 247                     .build();
 248         } catch (EntityNotFoundException e) {
 249             logger.warn("Not found: {}", e.getMessage());
 250             return null;
 251         }
 252     }
 253
 254     private String getPath(Policy policy) {
 255         return getPath(policy.getRic()) + "/" + policy.getId() + ".json";
 256     }
 257
 258     private String getPath(Ric ric) {
 259         return ric.id();
 260     }
 261
 262 }