Merge pull request #1298 from cushind/cushind/adding-shards-filtertype

cushind/adding-shards-filtertype

curator/actions.py

@@ -2088,6 +2088,7 @@ def do_dry_run(self):
         Show what a regular run would do, but don't actually do it.
         """
         self.index_list.filter_closed()
+        self.index_list.filter_by_shards(number_of_shards=self.number_of_shards)
         self.index_list.empty_list_check()
         try:
             index_lists = utils.chunk_index_list(self.index_list.indices)
@@ -2109,6 +2110,7 @@ def do_dry_run(self):
 
     def do_action(self):
         self.index_list.filter_closed()
+        self.index_list.filter_by_shards(number_of_shards=self.number_of_shards)
         self.index_list.empty_list_check()
         try:
             index_lists = utils.chunk_index_list(self.index_list.indices)

curator/defaults/filter_elements.py

@@ -81,6 +81,11 @@ def max_num_segments(**kwargs):
         Required('max_num_segments'): All(Coerce(int), Range(min=1))
     }
 
+def number_of_shards(**kwargs):
+    return {
+        Required('number_of_shards'): All(Coerce(int), Range(min=1))
+    }
+
 def pattern(**kwargs):
     return {
         Optional('pattern'): Any(*string_types)
@@ -102,6 +107,11 @@ def reverse(**kwargs):
     # Should be ignored if `use_age` is True
     return { Optional('reverse', default=True): Any(bool, All(Any(*string_types), Boolean())) }
 
+def shard_filter_behavior(**kwargs):
+    # This setting is only used with the shards filtertype and defaults to 'greater_than'.
+    return { Optional('shard_filter_behavior', default='greater_than'): 
+        Any('greater_than', 'less_than', 'greater_than_or_equal', 'less_than_or_equal', 'equal') }
+
 def source(**kwargs):
     # This setting is only used with the age filtertype, or with the space
     # filtertype when use_age is set to True.

curator/defaults/filtertypes.py

@@ -139,6 +139,13 @@ def state(action, config):
         filter_elements.exclude(),
     ]
 
+def shards(action, config):
+    return [
+        filter_elements.number_of_shards(),
+        filter_elements.shard_filter_behavior(),
+        filter_elements.exclude(),
+    ]
+
 def empty(action, config):
     return [
         filter_elements.exclude(),

curator/defaults/settings.py

@@ -81,6 +81,7 @@ def index_filtertypes():
         'pattern',
         'period',
         'space',
+        'shards'
     ]
 
 def snapshot_filtertypes():
@@ -120,11 +121,13 @@ def structural_filter_elements():
         Optional('key'): Any(*string_types),
         Optional('kind'): Any(*string_types),
         Optional('max_num_segments'): Coerce(int),
+        Optional('number_of_shards'): Coerce(int),
         Optional('pattern'): Any(*string_types),
         Optional('period_type'): Any(*string_types),
         Optional('reverse'): Any(None, bool, int, *string_types),
         Optional('range_from'): Coerce(int),
         Optional('range_to'): Coerce(int),
+        Optional('shard_filter_behavior'): Any(*string_types),
         Optional('source'): Any(*string_types),
         Optional('state'): Any(*string_types),
         Optional('stats_result'): Any(None, *string_types),

curator/indexlist.py

@@ -105,6 +105,7 @@ def __map_method(self, ft):
             'period': self.filter_period,
             'pattern': self.filter_by_regex,
             'space': self.filter_by_space,
+            'shards': self.filter_by_shards,
         }
         return methods[ft]
 
@@ -990,6 +991,55 @@ def filter_by_count(
                 self.__excludify(condition, exclude, index, msg)
                 idx += 1
 
+    def filter_by_shards(self, number_of_shards=None, shard_filter_behavior='greater_than', exclude=False):
+        """
+        Match `indices` with a given shard count.
+
+        Selects all indices with a shard count 'greater_than' number_of_shards by default.
+        Use shard_filter_behavior to select indices with shard count 'greater_than', 'greater_than_or_equal', 
+        'less_than', 'less_than_or_equal', or 'equal' to number_of_shards.
+
+        :arg number_of_shards: shard threshold 
+        :arg shard_filter_behavior: Do you want to filter on greater_than, greater_than_or_equal, less_than, 
+            less_than_or_equal, or equal?
+        :arg exclude: If `exclude` is `True`, this filter will remove matching
+            indices from `indices`. If `exclude` is `False`, then only matching
+            indices will be kept in `indices`.
+            Default is `False`
+        """
+        self.loggit.debug("Filtering indices by number of shards")
+        if not number_of_shards:
+            raise exceptions.MissingArgument('No value for "number_of_shards" provided')
+
+        if shard_filter_behavior not in ['greater_than', 'less_than', 'greater_than_or_equal', 'less_than_or_equal', 'equal']:
+            raise ValueError(
+                'Invalid value for "shard_filter_behavior": {0}'.format(
+                    shard_filter_behavior)
+            )
+
+        if number_of_shards < 1 or (shard_filter_behavior == 'less_than' and number_of_shards == 1):
+            raise ValueError(
+                'Unacceptable value: {0} -- "number_of_shards" cannot be less than 1. A valid index '
+                'will have at least one shard.'.format(number_of_shards)
+            )
+
+        self.empty_list_check()
+        for index in self.working_list():
+            self.loggit.debug('Filter by number of shards: Index: {0}'.format(index))
+
+            if shard_filter_behavior == 'greater_than':
+                condition = int(self.index_info[index]['number_of_shards']) > number_of_shards 
+            elif shard_filter_behavior == 'less_than':
+                condition = int(self.index_info[index]['number_of_shards']) < number_of_shards 
+            elif shard_filter_behavior == 'greater_than_or_equal':
+                condition = int(self.index_info[index]['number_of_shards']) >= number_of_shards 
+            elif shard_filter_behavior == 'less_than_or_equal':
+                condition = int(self.index_info[index]['number_of_shards']) <= number_of_shards 
+            else:
+                condition = int(self.index_info[index]['number_of_shards']) == number_of_shards 
+
+            self.__excludify(condition, exclude, index)
+
     def filter_period(
         self, period_type='relative', source='name', range_from=None, range_to=None,
         date_from=None, date_to=None, date_from_format=None, date_to_format=None,

test/integration/test_shrink.py

@@ -101,6 +101,26 @@
 '        kind: prefix\n'
 '        value: my\n')
 
+shrink_filter_by_shards = ('---\n'
+'actions:\n'
+'  1:\n'
+'    description: "Act on indices as filtered"\n'
+'    action: shrink\n'
+'    options:\n'
+'      shrink_node: {0}\n'
+'      node_filters:\n'
+'          {1}: {2}\n'
+'      number_of_shards: {3}\n'
+'      number_of_replicas: {4}\n'
+'      shrink_prefix: {5}\n'
+'      shrink_suffix: {6}\n'
+'      delete_after: {7}\n'
+'      wait_for_rebalance: {8}\n'
+'    filters:\n'
+'      - filtertype: shards\n'
+'        number_of_shards: {9}\n'
+'        shard_filter_behavior: {10}\n')
+
 class TestActionFileShrink(CuratorTestCase):
     def builder(self, action_args):
         self.idx = 'my_index'
@@ -247,7 +267,55 @@ def test_shrink_without_rebalance(self):
         indices = curator.get_indices(self.client)
         self.assertEqual(1, len(indices)) # Should only have `my_index-shrunken`
         self.assertEqual(indices[0], self.target)
+    def test_shrink_implicit_shard_filter(self):
+        self.create_index('my_invalid_shrink_index', shards=1)
+        self.create_index('my_valid_shrink_index', shards=5)
+
+        suffix = '-shrunken'
+        self.builder(
+            shrink.format(
+                'DETERMINISTIC',
+                'permit_masters',
+                'True',
+                1,
+                0,
+                '',
+                suffix,
+                'True',
+                'False'
+            )
+        )
+
+        indices = curator.get_indices(self.client)
+        self.assertEqual(3, len(indices)) 
+        self.assertTrue('my_invalid_shrink_index-shrunken' not in indices)
+        self.assertTrue('my_valid_shrink_index-shrunken' in indices)
+    def test_shrink_explicit_shard_filter(self):
+        self.create_index('my_invalid_shrink_index', shards=3)
+        self.create_index('my_valid_shrink_index', shards=5)
+
+        suffix = '-shrunken'
+        self.builder(
+            shrink_filter_by_shards.format(
+                'DETERMINISTIC',
+                'permit_masters',
+                'True',
+                1,
+                0,
+                '',
+                suffix,
+                'True',
+                'False',
+                5,
+                'greater_than_or_equal'
+            )
+        )
 
+        indices = curator.get_indices(self.client)
+        self.assertEqual(3, len(indices)) 
+        self.assertTrue('my_invalid_shrink_index-shrunken' not in indices)
+        self.assertTrue('my_valid_shrink_index-shrunken' in indices)
+        self.assertTrue('my_index-shrunken' not in indices)
 
 class TestCLIShrink(CuratorTestCase):
     def builder(self):

test/unit/test_class_index_list.py

@@ -1003,6 +1003,73 @@ def test_pattern_multiple_regex_groups(self):
         self.assertRaises(curator.ActionError, self.il.filter_by_count,
             count=1, use_age=True, pattern=r'^(\ )foo(\ )$', source='name', timestring='%Y.%m.%d',
         )
+
+class TestIndexListFilterShards(TestCase):
+    def builder(self):
+        self.client = Mock()
+        self.client.info.return_value = {'version': {'number': '5.0.0'} }
+        self.client.indices.get_settings.return_value = testvars.settings_two
+        self.client.cluster.state.return_value = testvars.clu_state_two
+        self.client.indices.stats.return_value = testvars.stats_two
+        self.il = curator.IndexList(self.client)
+    def test_filter_shards_raise(self):
+        self.builder()
+        self.assertRaises(curator.MissingArgument, self.il.filter_by_shards)
+    def test_bad_shard_count_raise_1(self):
+        self.builder()
+        self.assertRaises(curator.MissingArgument, self.il.filter_by_shards, number_of_shards=0)
+    def test_bad_shard_count_raise_2(self):
+        self.builder()
+        self.assertRaises(ValueError, self.il.filter_by_shards, number_of_shards=1, shard_filter_behavior='less_than')
+    def test_bad_shard_count_raise_3(self):
+        self.builder()
+        self.assertRaises(ValueError, self.il.filter_by_shards, number_of_shards=-1, shard_filter_behavior='greater_than')
+    def test_greater_than_or_equal(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='greater_than_or_equal')
+        self.assertEqual(
+            sorted([u'index-2016.03.03', u'index-2016.03.04']), sorted(self.il.indices))
+    def test_greater_than_or_equal_exclude(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='greater_than_or_equal', exclude=True)
+        self.assertEqual(
+            sorted([]), sorted(self.il.indices))
+    def test_greater_than(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5)
+        self.assertEqual(
+            sorted([]), sorted(self.il.indices))
+    def test_greater_than_exclude(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, exclude=True)
+        self.assertEqual(
+            sorted([u'index-2016.03.03', u'index-2016.03.04']), sorted(self.il.indices))
+    def test_less_than_or_equal(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='less_than_or_equal')
+        self.assertEqual(
+            sorted([u'index-2016.03.03', u'index-2016.03.04']), sorted(self.il.indices))
+    def test_less_than_or_equal_exclude(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='less_than_or_equal', exclude=True)
+        self.assertEqual(
+            sorted([]), sorted(self.il.indices))
+    def test_less_than(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='less_than')
+        self.assertEqual(
+            sorted([]), sorted(self.il.indices))
+    def test_less_than_exclude(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='less_than', exclude=True)
+        self.assertEqual(
+            sorted([u'index-2016.03.03', u'index-2016.03.04']), sorted(self.il.indices))
+    def test_equal(self):
+        self.builder()
+        self.il.filter_by_shards(number_of_shards=5, shard_filter_behavior='equal')
+        self.assertEqual(
+            sorted([u'index-2016.03.03', u'index-2016.03.04']), sorted(self.il.indices))
+
 class TestIndexListPeriodFilterName(TestCase):
     def test_get_name_based_age_in_range(self):
         unit = 'days'

test/unit/test_validators.py

@@ -220,3 +220,14 @@ def test_state(self):
             }
         ]
         self.assertEqual(config, shared_result(config, action))
+    def test_shards(self):
+        action = 'shrink'
+        config = [
+            {
+                'filtertype' : 'shards',
+                'number_of_shards' : 5,
+                'shard_filter_behavior': 'greater_than',
+                'exclude' : False,
+            }
+        ]
+        self.assertEqual(config, shared_result(config, action))