groupbasedpolicy/src/main/java/org/opendaylight/groupbasedpolicy/resolver/PolicyResolver.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.groupbasedpolicy.resolver;
   9
  10 import java.util.Collection;
  11 import java.util.HashSet;
  12 import java.util.List;
  13 import java.util.Set;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import java.util.concurrent.ConcurrentMap;
  16 import java.util.concurrent.atomic.AtomicReference;
  17
  18 import javax.annotation.concurrent.Immutable;
  19
  20 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  21 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  22 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  23 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  24 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  25 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  26 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
  27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  28 import org.opendaylight.groupbasedpolicy.api.PolicyValidatorRegistry;
  29 import org.opendaylight.groupbasedpolicy.api.ValidationResult;
  30 import org.opendaylight.groupbasedpolicy.api.Validator;
  31 import org.opendaylight.groupbasedpolicy.dto.EgKey;
  32 import org.opendaylight.groupbasedpolicy.dto.IndexedTenant;
  33 import org.opendaylight.groupbasedpolicy.dto.Policy;
  34 import org.opendaylight.groupbasedpolicy.util.DataStoreHelper;
  35 import org.opendaylight.groupbasedpolicy.util.InheritanceUtils;
  36 import org.opendaylight.groupbasedpolicy.util.PolicyInfoUtils;
  37 import org.opendaylight.groupbasedpolicy.util.PolicyResolverUtils;
  38 import org.opendaylight.groupbasedpolicy.util.TenantUtils;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.common.rev140421.ActionDefinitionId;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.common.rev140421.ClassifierDefinitionId;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.common.rev140421.TenantId;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.Tenant;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.tenant.policy.SubjectFeatureInstances;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.tenant.policy.subject.feature.instances.ActionInstance;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.tenant.policy.subject.feature.instances.ClassifierInstance;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.resolved.policy.rev150828.ResolvedPolicies;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.resolved.policy.rev150828.ResolvedPoliciesBuilder;
  48 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  49 import org.opendaylight.yangtools.yang.binding.DataObject;
  50 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 import com.google.common.base.Optional;
  55 import com.google.common.collect.HashMultimap;
  56 import com.google.common.collect.Multimaps;
  57 import com.google.common.collect.SetMultimap;
  58 import com.google.common.collect.Table;
  59 import com.google.common.util.concurrent.FutureCallback;
  60 import com.google.common.util.concurrent.Futures;
  61 import com.google.common.util.concurrent.ListenableFuture;
  62
  63 /**
  64  * The policy resolver is a utility for renderers to help in resolving
  65  * group-based policy into a form that is easier to apply to the actual network.
  66  *
  67  * For any pair of endpoint groups, there is a set of rules that could apply to
  68  * the endpoints on that group based on the policy configuration. The exact list
  69  * of rules that apply to a given pair of endpoints depends on the conditions
  70  * that are active on the endpoints.
  71  *
  72  * We need to be able to query against this policy model, enumerate the relevant
  73  * classes of traffic and endpoints, and notify renderers when there are changes
  74  * to policy as it applies to active sets of endpoints and endpoint groups.
  75  *
  76  * The policy resolver will maintain the necessary state for all tenants in its
  77  * control domain, which is the set of tenants for which policy listeners have
  78  * been registered.
  79  *
  80  */
  81 public class PolicyResolver implements PolicyValidatorRegistry, AutoCloseable {
  82
  83     private static final Logger LOG = LoggerFactory.getLogger(PolicyResolver.class);
  84
  85     private final DataBroker dataProvider;
  86
  87     private final FollowedTenantListener followedTenantListener;
  88
  89
  90     protected final ConcurrentMap<TenantId, TenantContext> resolvedTenants;
  91
  92     /*
  93      * Store validators for ActionDefinitions from Renderers
  94      *
  95      */
  96     private SetMultimap<ActionDefinitionId, Validator<ActionInstance>> actionInstanceValidatorsByDefinition = Multimaps.synchronizedSetMultimap(HashMultimap.<ActionDefinitionId, Validator<ActionInstance>>create());
  97     private SetMultimap<ClassifierDefinitionId, Validator<ClassifierInstance>> classifierInstanceValidatorsByDefinition = Multimaps.synchronizedSetMultimap(HashMultimap.<ClassifierDefinitionId, Validator<ClassifierInstance>>create());
  98
  99     public PolicyResolver(DataBroker dataProvider) {
 100         this.dataProvider = dataProvider;
 101         followedTenantListener = new FollowedTenantListener(dataProvider, this);
 102         resolvedTenants = new ConcurrentHashMap<>();
 103         LOG.debug("Initialized renderer common policy resolver");
 104     }
 105
 106     // *************
 107     // AutoCloseable
 108     // *************
 109     @Override
 110     public void close() throws Exception {
 111         for (TenantContext ctx : resolvedTenants.values()) {
 112             if (ctx.registration != null) {
 113                 ctx.registration.close();
 114             }
 115         }
 116         if (followedTenantListener != null) {
 117             followedTenantListener.close();
 118         }
 119     }
 120
 121     // *************************
 122     // PolicyResolutionValidatorRegistrar
 123     // *************************
 124
 125     @Override
 126     public void register(ActionDefinitionId actionDefinitionId,
 127             Validator<ActionInstance> validator) {
 128         actionInstanceValidatorsByDefinition.put(actionDefinitionId, validator);
 129     }
 130
 131     @Override
 132     public void unregister(ActionDefinitionId actionDefinitionId,
 133             Validator<ActionInstance> validator) {
 134         actionInstanceValidatorsByDefinition.remove(actionDefinitionId, validator);
 135     }
 136
 137     @Override
 138     public void register(ClassifierDefinitionId classifierDefinitionId,
 139             Validator<ClassifierInstance> validator) {
 140         classifierInstanceValidatorsByDefinition.put(classifierDefinitionId, validator);
 141     }
 142
 143     @Override
 144     public void unregister(ClassifierDefinitionId classifierDefinitionId,
 145             Validator<ClassifierInstance> validator) {
 146         classifierInstanceValidatorsByDefinition.remove(classifierDefinitionId, validator);
 147     }
 148
 149     /**
 150      * Subscribe the resolver to updates related to a particular tenant Make
 151      * sure that this can't be called concurrently with subscribe
 152      *
 153      * @param tenantId the tenant ID to subscribe to
 154      */
 155     protected void subscribeTenant(TenantId tenantId) {
 156         if (!resolvedTenants.containsKey(tenantId)) {
 157             updateTenant(tenantId);
 158         }
 159     }
 160
 161     /**
 162      * Unsubscribe the resolver from updates related to a particular tenant Make
 163      * sure that this can't be called concurrently with subscribe
 164      *
 165      * @param tenantId the tenant ID to subscribe to
 166      */
 167     protected void unsubscribeTenant(TenantId tenantId) {
 168         TenantContext context = resolvedTenants.get(tenantId);
 169         if (context != null) {
 170             resolvedTenants.remove(tenantId);
 171             context.registration.close();
 172         }
 173     }
 174
 175     private void updateTenant(final TenantId tenantId) {
 176         if (dataProvider == null) {
 177             return;
 178         }
 179
 180         TenantContext context = resolvedTenants.get(tenantId);
 181         if (context == null) {
 182             ListenerRegistration<DataChangeListener> registration = null;
 183             registration = dataProvider
 184                     .registerDataChangeListener(LogicalDatastoreType.CONFIGURATION,
 185                             TenantUtils.tenantIid(tenantId),
 186                             new PolicyChangeListener(tenantId),
 187                             DataChangeScope.SUBTREE);
 188             LOG.debug("Data change listener for tenant {} in CONF DS is registered.", tenantId.getValue());
 189
 190             context = new TenantContext(registration);
 191             TenantContext oldContext
 192                     = resolvedTenants.putIfAbsent(tenantId, context);
 193             if (oldContext != null) {
 194                 // already registered in a different thread; just use the other
 195                 // context
 196                 registration.close();
 197                 context = oldContext;
 198             } else {
 199                 LOG.info("Added tenant {} to policy scope", tenantId);
 200             }
 201         }
 202
 203         // Resolve the new tenant and update atomically
 204         final AtomicReference<IndexedTenant> tenantRef = context.tenant;
 205         final IndexedTenant ot = tenantRef.get();
 206         ReadOnlyTransaction transaction
 207                 = dataProvider.newReadOnlyTransaction();
 208         final InstanceIdentifier<Tenant> tiid = TenantUtils.tenantIid(tenantId);
 209         ListenableFuture<Optional<Tenant>> unresolved;
 210
 211         unresolved = transaction.read(LogicalDatastoreType.CONFIGURATION, tiid);
 212
 213         Futures.addCallback(unresolved, new FutureCallback<Optional<Tenant>>() {
 214             @Override
 215             public void onSuccess(Optional<Tenant> result) {
 216                 if (!result.isPresent()) {
 217                     LOG.info("Tenant {} not found in CONF; check&delete from OPER", tenantId);
 218                     deleteOperTenantIfExists(tiid, tenantId);
 219                     return;
 220                 }
 221                 LOG.debug("Resolving of tenant inheritance and policy triggered by a change in tenant {}", tenantId);
 222                 Tenant t = InheritanceUtils.resolveTenant(result.get());
 223                 if (t.getPolicy() != null && t.getPolicy().getSubjectFeatureInstances() != null) {
 224                     SubjectFeatureInstances subjectFeatureInstances = t.getPolicy().getSubjectFeatureInstances();
 225                     // TODO log and remove invalid action instances
 226                     if (actionInstancesAreValid(subjectFeatureInstances.getActionInstance())
 227                             && classifierInstancesAreValid(subjectFeatureInstances.getClassifierInstance())) {
 228                     IndexedTenant it = new IndexedTenant(t);
 229                     if (!tenantRef.compareAndSet(ot, it)) {
 230                         // concurrent update of tenant policy. Retry
 231                         updateTenant(tenantId);
 232                     } else {
 233                         // Update the policy cache and notify listeners
 234                         WriteTransaction wt = dataProvider.newWriteOnlyTransaction();
 235                         wt.put(LogicalDatastoreType.OPERATIONAL, tiid, t, true);
 236                         wt.submit();
 237                         updatePolicy();
 238                     }
 239                 }
 240             }
 241         }
 242
 243             @Override
 244             public void onFailure(Throwable t) {
 245                 LOG.error("Count not get tenant {}", tenantId, t);
 246             }
 247         });
 248     }
 249
 250     private void deleteOperTenantIfExists(final InstanceIdentifier<Tenant> tiid, final TenantId tenantId) {
 251         final ReadWriteTransaction rwTx = dataProvider.newReadWriteTransaction();
 252
 253         ListenableFuture<Optional<Tenant>> readFuture = rwTx.read(LogicalDatastoreType.OPERATIONAL, tiid);
 254         Futures.addCallback(readFuture, new FutureCallback<Optional<Tenant>>() {
 255             @Override
 256             public void onSuccess(Optional<Tenant> result) {
 257                 if(result.isPresent()){
 258                     TenantContext tenantContext = resolvedTenants.get(tenantId);
 259                     tenantContext.tenant.set(null);
 260                     rwTx.delete(LogicalDatastoreType.OPERATIONAL, tiid);
 261                     rwTx.submit();
 262                     updatePolicy();
 263                 }
 264             }
 265
 266             @Override
 267             public void onFailure(Throwable t) {
 268                 LOG.error("Failed to read operational datastore: {}", t);
 269                 rwTx.cancel();
 270             }
 271         });
 272     }
 273
 274     private void updatePolicy() {
 275         try {
 276             Set<IndexedTenant> indexedTenants = getIndexedTenants(resolvedTenants.values());
 277             Table<EgKey, EgKey, Policy> policyMap = PolicyResolverUtils.resolvePolicy(indexedTenants);
 278             updatePolicyInDataStore(policyMap);
 279         } catch (Exception e) {
 280             LOG.error("Failed to update policy", e);
 281         }
 282     }
 283
 284     private void updatePolicyInDataStore(Table<EgKey, EgKey, Policy> policyMap) {
 285         if (dataProvider == null) {
 286             LOG.error("Couldn't Write Resolved Tenants Policy Info to Datastore because dataProvider is NULL");
 287             return;
 288         }
 289         ResolvedPolicies resolvedPolicies = new ResolvedPoliciesBuilder().setResolvedPolicy(
 290                 PolicyInfoUtils.buildResolvedPolicy(policyMap)).build();
 291
 292         WriteTransaction t = dataProvider.newWriteOnlyTransaction();
 293         t.put(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.builder(ResolvedPolicies.class).build(),
 294                 resolvedPolicies, true);
 295         if (DataStoreHelper.submitToDs(t)) {
 296             LOG.debug("Wrote resolved policies to Datastore");
 297         } else {
 298             LOG.error("Failed to write resolved policies to Datastore.");
 299         }
 300     }
 301
 302     private Set<IndexedTenant> getIndexedTenants(Collection<TenantContext> tenantCtxs) {
 303         Set<IndexedTenant> result = new HashSet<>();
 304         for (TenantContext tenant : tenantCtxs) {
 305             IndexedTenant t = tenant.tenant.get();
 306             if (t != null) {
 307                 result.add(t);
 308             }
 309         }
 310         return result;
 311     }
 312
 313     /**
 314      * Validation of action instances.
 315      *
 316      * @param actionInstances list of instances to validate
 317      * @return true if instances are valid or if <code>actionInstances</code>
 318      * is <code>null</code>, Otherwise returns false.
 319      *
 320      */
 321     private boolean actionInstancesAreValid(List<ActionInstance> actionInstances) {
 322         if (actionInstances == null) {
 323             return true;
 324         }
 325         for (ActionInstance actionInstance : actionInstances) {
 326             Set<Validator<ActionInstance>> actionInstanceValidators = actionInstanceValidatorsByDefinition.get(actionInstance.getActionDefinitionId());
 327             for (Validator<ActionInstance> actionInstanceValidator : actionInstanceValidators) {
 328                 ValidationResult validationResult = actionInstanceValidator.validate(actionInstance);
 329                 if (!validationResult.isValid()) {
 330                     LOG.error("ActionInstance {} is not valid! {}", actionInstance.getName().getValue(),
 331                             validationResult.getMessage());
 332                     return false;
 333                 }
 334             }
 335         }
 336         return true;
 337     }
 338
 339     /**
 340      * Validation of classifier instances.
 341      *
 342      * @param classifierInstances list of instances to validate
 343      * @return true if instances are valid or if <code>classifierInstances</code>
 344      * is <code>null</code>, Otherwise returns false.
 345      *
 346      */
 347     private boolean classifierInstancesAreValid(List<ClassifierInstance> classifierInstances) {
 348         if (classifierInstances == null) {
 349             return true;
 350         }
 351         for (ClassifierInstance classifierInstance : classifierInstances) {
 352             Set<Validator<ClassifierInstance>> classifierInstanceValidators = classifierInstanceValidatorsByDefinition.get(classifierInstance.getClassifierDefinitionId());
 353             for (Validator<ClassifierInstance> classifierInstanceValidator : classifierInstanceValidators) {
 354                 ValidationResult validationResult = classifierInstanceValidator.validate(classifierInstance);
 355                 if (!validationResult.isValid()) {
 356                     LOG.error("ClassifierInstance {} is not valid! {}", classifierInstance.getName().getValue(),
 357                             validationResult.getMessage());
 358                     return false;
 359                 }
 360             }
 361         }
 362         return true;
 363     }
 364
 365     static class TenantContext {
 366
 367         final ListenerRegistration<DataChangeListener> registration;
 368
 369         AtomicReference<IndexedTenant> tenant = new AtomicReference<>();
 370
 371         public TenantContext(ListenerRegistration<DataChangeListener> registration) {
 372             super();
 373             this.registration = registration;
 374         }
 375     }
 376
 377     @Immutable
 378     private class PolicyChangeListener implements DataChangeListener {
 379
 380         final TenantId tenantId;
 381
 382         public PolicyChangeListener(TenantId tenantId) {
 383             super();
 384             this.tenantId = tenantId;
 385         }
 386
 387         @Override
 388         public void onDataChanged(AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> arg0) {
 389             updateTenant(tenantId);
 390         }
 391
 392     }
 393
 394 }