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         Policy previousDef = this.get(policy.getId());
 101         if (previousDef != null) {
 102             removeFromMaps(previousDef);
 103         }
 104
 105         policiesId.put(policy.getId(), policy);
 106         policiesRic.put(policy.getRic().id(), policy.getId(), policy);
 107         policiesService.put(policy.getOwnerServiceId(), policy.getId(), policy);
 108         policiesType.put(policy.getType().getId(), policy.getId(), policy);
 109         if (!policy.isTransient()) {
 110             store(policy);
 111         }
 112     }
 113
 114     public synchronized boolean containsPolicy(String id) {
 115         return policiesId.containsKey(id);
 116     }
 117
 118     public synchronized Policy get(String id) {
 119         return policiesId.get(id);
 120     }
 121
 122     public synchronized Policy getPolicy(String id) throws EntityNotFoundException {
 123         Policy p = policiesId.get(id);
 124         if (p == null) {
 125             throw new EntityNotFoundException("Could not find policy: " + id);
 126         }
 127         return p;
 128     }
 129
 130     public synchronized Collection<Policy> getAll() {
 131         return new Vector<>(policiesId.values());
 132     }
 133
 134     public synchronized Collection<Policy> getForService(String service) {
 135         return policiesService.get(service);
 136     }
 137
 138     public synchronized Collection<Policy> getForRic(String ric) {
 139         return policiesRic.get(ric);
 140     }
 141
 142     public synchronized Collection<Policy> getForType(String type) {
 143         return policiesType.get(type);
 144     }
 145
 146     public synchronized Policy removeId(String id) {
 147         Policy p = policiesId.get(id);
 148         if (p != null) {
 149             remove(p);
 150         }
 151         return p;
 152     }
 153
 154     public synchronized void remove(Policy policy) {
 155         if (!policy.isTransient()) {
 156             dataStore.deleteObject(getPath(policy)).subscribe();
 157         }
 158         removeFromMaps(policy);
 159     }
 160
 161     public synchronized void removePoliciesForRic(String ricId) {
 162         Collection<Policy> policiesForRic = getForRic(ricId);
 163         for (Policy policy : policiesForRic) {
 164             remove(policy);
 165         }
 166     }
 167
 168     public Collection<Policy> filterPolicies(@Nullable String typeId, @Nullable String ricId,
 169             @Nullable String serviceId, @Nullable String typeName) {
 170
 171         if (typeId != null) {
 172             return filter(this.getForType(typeId), null, ricId, serviceId, typeName);
 173         } else if (serviceId != null) {
 174             return filter(this.getForService(serviceId), typeId, ricId, null, typeName);
 175         } else if (ricId != null) {
 176             return filter(this.getForRic(ricId), typeId, null, serviceId, typeName);
 177         } else {
 178             return filter(this.getAll(), typeId, ricId, serviceId, typeName);
 179         }
 180     }
 181
 182     public synchronized int size() {
 183         return policiesId.size();
 184     }
 185
 186     public synchronized void clear() {
 187         while (policiesId.size() > 0) {
 188             Set<String> keys = policiesId.keySet();
 189             removeId(keys.iterator().next());
 190         }
 191         dataStore.deleteAllObjects().onErrorResume(t -> Mono.empty()).subscribe();
 192     }
 193
 194     private void store(Policy policy) {
 195
 196         byte[] bytes = gson.toJson(toStorageObject(policy)).getBytes();
 197         this.dataStore.writeObject(this.getPath(policy), bytes) //
 198                 .doOnError(t -> logger.error("Could not store job in S3, reason: {}", t.getMessage())) //
 199                 .subscribe();
 200     }
 201
 202     private void removeFromMaps(Policy policy) {
 203         policiesId.remove(policy.getId());
 204         policiesRic.remove(policy.getRic().id(), policy.getId());
 205         policiesService.remove(policy.getOwnerServiceId(), policy.getId());
 206         policiesType.remove(policy.getType().getId(), policy.getId());
 207     }
 208
 209     private boolean isMatch(String filterValue, String actualValue) {
 210         return filterValue == null || actualValue.equals(filterValue);
 211     }
 212
 213     private boolean isTypeMatch(Policy policy, @Nullable String typeName) {
 214         return (typeName == null) || policy.getType().getTypeId().getName().equals(typeName);
 215     }
 216
 217     private Collection<Policy> filter(Collection<Policy> collection, String typeId, String ricId, String serviceId,
 218             String typeName) {
 219         if (typeId == null && ricId == null && serviceId == null && typeName == null) {
 220             return collection;
 221         }
 222         List<Policy> filtered = new ArrayList<>(collection.size());
 223         for (Policy p : collection) {
 224             if (isMatch(typeId, p.getType().getId()) && isMatch(ricId, p.getRic().id())
 225                     && isMatch(serviceId, p.getOwnerServiceId()) && isTypeMatch(p, typeName)) {
 226                 filtered.add(p);
 227             }
 228         }
 229         return filtered;
 230     }
 231
 232     private PersistentPolicyInfo toStorageObject(Policy p) {
 233         return PersistentPolicyInfo.builder() //
 234                 .id(p.getId()) //
 235                 .json(p.getJson()) //
 236                 .ownerServiceId(p.getOwnerServiceId()) //
 237                 .ricId(p.getRic().id()) //
 238                 .statusNotificationUri(p.getStatusNotificationUri()) //
 239                 .typeId(p.getType().getId()) //
 240                 .isTransient(p.isTransient()) //
 241                 .lastModified(p.getLastModified().toString()) //
 242                 .build();
 243     }
 244
 245     private Policy toPolicy(PersistentPolicyInfo p, Ric ric, PolicyTypes types) {
 246         try {
 247             return Policy.builder() //
 248                     .id(p.getId()) //
 249                     .isTransient(p.isTransient()) //
 250                     .json(p.getJson()) //
 251                     .lastModified(Instant.parse(p.lastModified)) //
 252                     .ownerServiceId(p.getOwnerServiceId()) //
 253                     .ric(ric) //
 254                     .statusNotificationUri(p.getStatusNotificationUri()) //
 255                     .type(types.getType(p.getTypeId())) //
 256                     .build();
 257         } catch (EntityNotFoundException e) {
 258             logger.warn("Not found: {}", e.getMessage());
 259             return null;
 260         }
 261     }
 262
 263     private String getPath(Policy policy) {
 264         return getPath(policy.getRic()) + "/" + policy.getId() + ".json";
 265     }
 266
 267     private String getPath(Ric ric) {
 268         return ric.id();
 269     }
 270
 271 }