metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/WindowLookbackTest.java - metron - Git at Google

 /*
  *
  *  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 org.apache.metron.profiler.client.stellar;

 import org.apache.commons.lang3.Range;
 import org.apache.metron.profiler.ProfilePeriod;
 import org.apache.metron.profiler.client.window.WindowProcessor;
 import org.apache.metron.stellar.common.StellarProcessor;
 import org.apache.metron.stellar.dsl.Context;
 import org.apache.metron.stellar.dsl.DefaultVariableResolver;
 import org.apache.metron.stellar.dsl.ParseException;
 import org.apache.metron.stellar.dsl.functions.resolver.FunctionResolver;
 import org.apache.metron.stellar.dsl.functions.resolver.SimpleFunctionResolver;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;

 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;

 import static org.junit.jupiter.api.Assertions.*;

 public class WindowLookbackTest {

   static FunctionResolver resolver;
   static Context context;
   @BeforeAll
   public static void setup() {
     resolver = new SimpleFunctionResolver()
                     .withClass(GetProfile.class)
                     .withClass(FixedLookback.class)
                     .withClass(WindowLookback.class);
     context = new Context.Builder()
                     .with(Context.Capabilities.GLOBAL_CONFIG, () -> new HashMap<>())
                     .build();
   }

   @Test
   public void testSpecifyingConfig() {
     //we should be able to specify the config and have it take hold.  If we change the
     //profile duration to 1 minute instead of 15 minutes (the default), then we should see
     //the correct number of profiles.
     long durationMs = 60000;

     Map<String, Object> config = new HashMap<>();
     config.put(ProfilerClientConfig.PROFILER_PERIOD.getKey(), 1);

     State state = test("1 hour", new Date(), Optional.of(config), Assertions.NOT_EMPTY, Assertions.CONTIGUOUS);
     assertEquals(TimeUnit.HOURS.toMillis(1) / durationMs, state.periods.size());
   }

   @Test
   public void testSpecifyingOnlySelector() {
     String stellarStatement = "PROFILE_WINDOW('1 hour')";
     Map<String, Object> variables = new HashMap<>();
     StellarProcessor stellar = new StellarProcessor();
     List<ProfilePeriod> periods = (List<ProfilePeriod>)stellar.parse( stellarStatement
                                                                     , new DefaultVariableResolver(k -> variables.get(k),k -> variables.containsKey(k))
                                                                     , resolver
                                                                     , context
                                                                     );
     assertEquals(TimeUnit.HOURS.toMillis(1) / getDurationMs(), periods.size());
   }

   @Test
   public void testDenseLookback() {
     State state = test("1 hour", Assertions.NOT_EMPTY, Assertions.CONTIGUOUS);
     assertEquals(TimeUnit.HOURS.toMillis(1) / getDurationMs(), state.periods.size());
   }

   @Test
   public void testShiftedDenseLookback() {
     State state = test("from 2 hours ago to 30 minutes ago", Assertions.NOT_EMPTY
                                                            , Assertions.CONTIGUOUS
                                                            , Assertions.INTERVALS_CONTAIN_ALL_PERIODS
                                                            );
     assertEquals(TimeUnit.MINUTES.toMillis(90) / getDurationMs(), state.periods.size());
   }

   @Test
   public void testShiftedSparseLookback() {
     State state = test("30 minute window every 1 hour from 2 hours ago to 30 minutes ago", Assertions.NOT_EMPTY
                                                                                          , Assertions.DISCONTIGUOUS
                                                                                          , Assertions.INTERVALS_CONTAIN_ALL_PERIODS
                                                                                          );
     assertEquals(TimeUnit.MINUTES.toMillis(60) / getDurationMs(), state.periods.size());
   }

   @Test
   public void testEmptyDueToExclusions() {
     test("30 minute window every 24 hours from 7 days ago including saturdays excluding weekends", Assertions.EMPTY);
   }

   @Test
   public void testErrorInSelector() {
     assertThrows(
         ParseException.class,
         () ->
             test(
                 "30 minute idow every 24 hours from 7 days ago including saturdays excluding weekends",
                 Assertions.EMPTY));
   }

   long getDurationMs() {
     int duration = ProfilerClientConfig.PROFILER_PERIOD.getDefault(Integer.class);
     TimeUnit unit = TimeUnit.valueOf(ProfilerClientConfig.PROFILER_PERIOD_UNITS.getDefault(String.class));
     return unit.toMillis(duration);
   }

   public State test(String windowSelector, Assertions... assertions) {
     return test(windowSelector, new Date(), Optional.empty(), assertions);
   }

   public State test(String windowSelector, Date now, Optional<Map<String, Object>> config, Assertions... assertions) {

     List<Range<Long>> windowIntervals = WindowProcessor.process(windowSelector).toIntervals(now.getTime());
     String stellarStatement = "PROFILE_WINDOW('" + windowSelector + "', now"
                             + (config.isPresent()?", config":"")
                             + ")";
     Map<String, Object> variables = new HashMap<>();
     variables.put("now", now.getTime());
     if(config.isPresent()) {
       variables.put("config", config.get());
     }
     StellarProcessor stellar = new StellarProcessor();
     List<ProfilePeriod> periods = (List<ProfilePeriod>)stellar.parse( stellarStatement
                                                                     , new DefaultVariableResolver(k -> variables.get(k),k -> variables.containsKey(k))
                                                                     , resolver
                                                                     , context
                                                                     );
     State state = new State(windowIntervals, periods);
     for(Assertions assertion : assertions) {
       assertTrue(assertion.test(state), assertion.name());
     }
     return state;
   }

   private enum Assertions implements Predicate<State>{
     EMPTY( state -> state.windowIntervals.isEmpty() && state.periods.isEmpty() ),
     NOT_EMPTY( state -> !state.windowIntervals.isEmpty() && !state.periods.isEmpty()),
     CONTIGUOUS( state -> {
       if(state.periods.size() < 2) {
         return true;
       }
       long duration = state.periods.get(1).getStartTimeMillis() - state.periods.get(0).getStartTimeMillis();
       for(int i = 1;i < state.periods.size();++i) {
         long left = state.periods.get(i - 1).getStartTimeMillis();
         long right = state.periods.get(i).getStartTimeMillis();
         if(right - left != duration) {
           return false;
         }
       }
       return true;
     }),
     DISCONTIGUOUS( state -> !Assertions.CONTIGUOUS.test(state)),
     INTERVALS_CONTAIN_ALL_PERIODS( state -> {
       List<Range<Long>> windowIntervals = state.windowIntervals;
       List<ProfilePeriod> periods = state.periods;

       Set<Range<Long>> foundIntervals = new HashSet<>();
       for(ProfilePeriod period : periods) {
         boolean found = false;
         for(Range<Long> interval : windowIntervals) {
           if(interval.contains(period.getStartTimeMillis())) {
             foundIntervals.add(interval);
             found = true;
           }
         }
         if(!found) {
           return false;
         }
       }
       return foundIntervals.size() == windowIntervals.size();
     })
     ;
     Predicate<State> predicate;
     Assertions(Predicate<State> predicate) {
       this.predicate = predicate;
     }

     @Override
     public boolean test(State s) {
       return predicate.test(s);
     }
   }

   private static class State {
     List<Range<Long>> windowIntervals;
     List<ProfilePeriod> periods;
     public State(List<Range<Long>> windowIntervals, List<ProfilePeriod> periods) {
       this.periods = periods;
       this.windowIntervals = windowIntervals;
     }
   }

 }
	/*
	*
	* 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 org.apache.metron.profiler.client.stellar;

	import org.apache.commons.lang3.Range;
	import org.apache.metron.profiler.ProfilePeriod;
	import org.apache.metron.profiler.client.window.WindowProcessor;
	import org.apache.metron.stellar.common.StellarProcessor;
	import org.apache.metron.stellar.dsl.Context;
	import org.apache.metron.stellar.dsl.DefaultVariableResolver;
	import org.apache.metron.stellar.dsl.ParseException;
	import org.apache.metron.stellar.dsl.functions.resolver.FunctionResolver;
	import org.apache.metron.stellar.dsl.functions.resolver.SimpleFunctionResolver;
	import org.junit.jupiter.api.BeforeAll;
	import org.junit.jupiter.api.Test;

	import java.util.*;
	import java.util.concurrent.TimeUnit;
	import java.util.function.Predicate;

	import static org.junit.jupiter.api.Assertions.*;

	public class WindowLookbackTest {

	static FunctionResolver resolver;
	static Context context;
	@BeforeAll
	public static void setup() {
	resolver = new SimpleFunctionResolver()
	.withClass(GetProfile.class)
	.withClass(FixedLookback.class)
	.withClass(WindowLookback.class);
	context = new Context.Builder()
	.with(Context.Capabilities.GLOBAL_CONFIG, () -> new HashMap<>())
	.build();
	}

	@Test
	public void testSpecifyingConfig() {
	//we should be able to specify the config and have it take hold. If we change the
	//profile duration to 1 minute instead of 15 minutes (the default), then we should see
	//the correct number of profiles.
	long durationMs = 60000;

	Map<String, Object> config = new HashMap<>();
	config.put(ProfilerClientConfig.PROFILER_PERIOD.getKey(), 1);

	State state = test("1 hour", new Date(), Optional.of(config), Assertions.NOT_EMPTY, Assertions.CONTIGUOUS);
	assertEquals(TimeUnit.HOURS.toMillis(1) / durationMs, state.periods.size());
	}

	@Test
	public void testSpecifyingOnlySelector() {
	String stellarStatement = "PROFILE_WINDOW('1 hour')";
	Map<String, Object> variables = new HashMap<>();
	StellarProcessor stellar = new StellarProcessor();
	List<ProfilePeriod> periods = (List<ProfilePeriod>)stellar.parse( stellarStatement
	, new DefaultVariableResolver(k -> variables.get(k),k -> variables.containsKey(k))
	, resolver
	, context
	);
	assertEquals(TimeUnit.HOURS.toMillis(1) / getDurationMs(), periods.size());
	}

	@Test
	public void testDenseLookback() {
	State state = test("1 hour", Assertions.NOT_EMPTY, Assertions.CONTIGUOUS);
	assertEquals(TimeUnit.HOURS.toMillis(1) / getDurationMs(), state.periods.size());
	}

	@Test
	public void testShiftedDenseLookback() {
	State state = test("from 2 hours ago to 30 minutes ago", Assertions.NOT_EMPTY
	, Assertions.CONTIGUOUS
	, Assertions.INTERVALS_CONTAIN_ALL_PERIODS
	);
	assertEquals(TimeUnit.MINUTES.toMillis(90) / getDurationMs(), state.periods.size());
	}

	@Test
	public void testShiftedSparseLookback() {
	State state = test("30 minute window every 1 hour from 2 hours ago to 30 minutes ago", Assertions.NOT_EMPTY
	, Assertions.DISCONTIGUOUS
	, Assertions.INTERVALS_CONTAIN_ALL_PERIODS
	);
	assertEquals(TimeUnit.MINUTES.toMillis(60) / getDurationMs(), state.periods.size());
	}

	@Test
	public void testEmptyDueToExclusions() {
	test("30 minute window every 24 hours from 7 days ago including saturdays excluding weekends", Assertions.EMPTY);
	}

	@Test
	public void testErrorInSelector() {
	assertThrows(
	ParseException.class,
	() ->
	test(
	"30 minute idow every 24 hours from 7 days ago including saturdays excluding weekends",
	Assertions.EMPTY));
	}

	long getDurationMs() {
	int duration = ProfilerClientConfig.PROFILER_PERIOD.getDefault(Integer.class);
	TimeUnit unit = TimeUnit.valueOf(ProfilerClientConfig.PROFILER_PERIOD_UNITS.getDefault(String.class));
	return unit.toMillis(duration);
	}

	public State test(String windowSelector, Assertions... assertions) {
	return test(windowSelector, new Date(), Optional.empty(), assertions);
	}

	public State test(String windowSelector, Date now, Optional<Map<String, Object>> config, Assertions... assertions) {

	List<Range<Long>> windowIntervals = WindowProcessor.process(windowSelector).toIntervals(now.getTime());
	String stellarStatement = "PROFILE_WINDOW('" + windowSelector + "', now"
	+ (config.isPresent()?", config":"")
	+ ")";
	Map<String, Object> variables = new HashMap<>();
	variables.put("now", now.getTime());
	if(config.isPresent()) {
	variables.put("config", config.get());
	}
	StellarProcessor stellar = new StellarProcessor();
	List<ProfilePeriod> periods = (List<ProfilePeriod>)stellar.parse( stellarStatement
	, new DefaultVariableResolver(k -> variables.get(k),k -> variables.containsKey(k))
	, resolver
	, context
	);
	State state = new State(windowIntervals, periods);
	for(Assertions assertion : assertions) {
	assertTrue(assertion.test(state), assertion.name());
	}
	return state;
	}

	private enum Assertions implements Predicate<State>{
	EMPTY( state -> state.windowIntervals.isEmpty() && state.periods.isEmpty() ),
	NOT_EMPTY( state -> !state.windowIntervals.isEmpty() && !state.periods.isEmpty()),
	CONTIGUOUS( state -> {
	if(state.periods.size() < 2) {
	return true;
	}
	long duration = state.periods.get(1).getStartTimeMillis() - state.periods.get(0).getStartTimeMillis();
	for(int i = 1;i < state.periods.size();++i) {
	long left = state.periods.get(i - 1).getStartTimeMillis();
	long right = state.periods.get(i).getStartTimeMillis();
	if(right - left != duration) {
	return false;
	}
	}
	return true;
	}),
	DISCONTIGUOUS( state -> !Assertions.CONTIGUOUS.test(state)),
	INTERVALS_CONTAIN_ALL_PERIODS( state -> {
	List<Range<Long>> windowIntervals = state.windowIntervals;
	List<ProfilePeriod> periods = state.periods;

	Set<Range<Long>> foundIntervals = new HashSet<>();
	for(ProfilePeriod period : periods) {
	boolean found = false;
	for(Range<Long> interval : windowIntervals) {
	if(interval.contains(period.getStartTimeMillis())) {
	foundIntervals.add(interval);
	found = true;
	}
	}
	if(!found) {
	return false;
	}
	}
	return foundIntervals.size() == windowIntervals.size();
	})
	;
	Predicate<State> predicate;
	Assertions(Predicate<State> predicate) {
	this.predicate = predicate;
	}

	@Override
	public boolean test(State s) {
	return predicate.test(s);
	}
	}

	private static class State {
	List<Range<Long>> windowIntervals;
	List<ProfilePeriod> periods;
	public State(List<Range<Long>> windowIntervals, List<ProfilePeriod> periods) {
	this.periods = periods;
	this.windowIntervals = windowIntervals;
	}
	}

	}