apps/logstream/src/main/java/com/datatorrent/apps/logstream/FilterOperator.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.datatorrent.apps.logstream;

import java.util.*;
import java.util.HashMap;
import java.util.Map;

import javax.validation.ValidationException;

import org.codehaus.janino.ExpressionEvaluator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datatorrent.api.*;
import com.datatorrent.api.Context.OperatorContext;

import com.datatorrent.apps.logstream.PropertyRegistry.LogstreamPropertyRegistry;
import com.datatorrent.apps.logstream.PropertyRegistry.PropertyRegistry;
import com.datatorrent.common.util.BaseOperator;
import com.datatorrent.netlet.util.DTThrowable;

/**
 *
 * Filters an input tuple and emits a filter stamped tuple for each satisfied filter
 *
 * @since 0.9.4
 */
public class FilterOperator extends BaseOperator
{
  public final transient DefaultOutputPort<HashMap<String, Object>> outputMap = new DefaultOutputPort<HashMap<String, Object>>();
  public final transient DefaultInputPort<Map<String, Object>> input = new DefaultInputPort<Map<String, Object>>()
  {
    @Override
    public void process(Map<String, Object> map)
    {
      HashMap<String, Object> filterTuple;
      int typeId = (Integer)map.get(LogstreamUtil.LOG_TYPE);
      Map<String, String[]> conditions = conditionList.get(typeId);
      if (conditions != null) {
        for (String condition : conditions.keySet()) {
          if (evaluate(condition, map, conditions.get(condition))) {
            int index = registry.getIndex(LogstreamUtil.FILTER, condition);
            filterTuple = new HashMap<String, Object>(map);
            filterTuple.put(LogstreamUtil.FILTER, index);
            outputMap.emit(filterTuple);
          }
        }
      }

      // emit the same tuple for default condition
      int defaultFilterIndex = registry.getIndex(LogstreamUtil.FILTER, registry.lookupValue(typeId) + "_" + "DEFAULT");

      if (defaultFilterIndex >= 0) {
        map.put(LogstreamUtil.FILTER, defaultFilterIndex);
        outputMap.emit((HashMap<String, Object>)map);
      }
    }

    private boolean evaluate(String condition, Map<String, Object> map, String[] keys)
    {
      boolean ret = false;

      Object[] values = new Object[keys.length];
      Class[] classTypes = new Class[keys.length];
      ExpressionEvaluator ee = evaluators.get(condition);

      try {
        for (int i = 0; i < keys.length; i++) {
          Object val = map.get(keys[i]);

          if (val == null) {
            logger.debug("filter key {} missing in input record {}", keys[i], map);
            return ret;
          }
          else {
            values[i] = val;
            classTypes[i] = val.getClass();
          }
        }

        if (ee == null) {
          ee = new ExpressionEvaluator(condition, Boolean.class, keys, classTypes);
          evaluators.put(condition, ee);
        }

        ret = (Boolean)ee.evaluate(values);
        logger.debug("expression evaluated to {} for expression: {} with key class types: {} keys: {} values: {}", ret, condition, Arrays.toString(classTypes), Arrays.toString(keys), Arrays.toString(values));

      }
      catch (Throwable t) {
        DTThrowable.rethrow(t);
      }

      return ret;
    }

  };
  private static final Logger logger = LoggerFactory.getLogger(FilterOperator.class);
  /**
   * key: type
   * value --> map of
   * key: condition expression
   * value: list of keys on which the condition is
   */
  private HashMap<Integer, Map<String, String[]>> conditionList = new HashMap<Integer, Map<String, String[]>>();
  private transient HashMap<String, ExpressionEvaluator> evaluators = new HashMap<String, ExpressionEvaluator>();
  private PropertyRegistry<String> registry;

  /**
   * supply the registry object which is used to store and retrieve meta information about each tuple
   *
   * @param registry
   */
  public void setRegistry(PropertyRegistry<String> registry)
  {
    this.registry = registry;
  }

  @Override
  public void setup(OperatorContext context)
  {
    super.setup(context);
    LogstreamPropertyRegistry.setInstance(registry);
  }

  /**
   * Supply the properties to the operator.
   * Input includes following properties:
   * type=logtype // input logtype for which the properties are to be set
   * followed by list of keys followed by the filter expression
   * eg: type=apache, response, response.equals("404")
   *
   * if default filter is needed i.e. emit a tuple with no filter applied on it
   * Input includes following properties:
   * type=logtype // input logtype for which the properties are to be set
   * followed by the following default condition
   * default=true
   * eg: type=apache, default=true
   *
   * @param properties
   */
  public void addFilterCondition(String[] properties)
  {
    try {
      // TODO: validations
      if (properties.length == 2) {
        logger.debug(Arrays.toString(properties));
        String[] split = properties[0].split("=");
        String type = split[1];
        String[] split1 = properties[1].split("=");
        if (split1[1].toLowerCase().equals("true")) {
          registry.bind(LogstreamUtil.FILTER, type + "_" + "DEFAULT");
        }
      }
      else if (properties.length > 2) {
        String[] split = properties[0].split("=");
        String type = split[1];
        int typeId = registry.getIndex(LogstreamUtil.LOG_TYPE, type);
        String[] keys = new String[properties.length - 2];

        System.arraycopy(properties, 1, keys, 0, keys.length);

        String expression = properties[properties.length - 1];

        Map<String, String[]> conditions = conditionList.get(typeId);
        if (conditions == null) {
          conditions = new HashMap<String, String[]>();
          conditionList.put(typeId, conditions);
        }

        conditions.put(expression, keys);
        if (registry != null) {
          registry.bind(LogstreamUtil.FILTER, expression);
        }
      }
      else {
        throw new ValidationException("Invalid input property string " + Arrays.toString(properties));
      }
    }
    catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }
}