Merge pull request #916 from untergeek/feature/751

Reindex is here!

.travis.yml

@@ -9,6 +9,7 @@ env:
   - ES_VERSION=5.0.2
   - ES_VERSION=5.1.2
   - ES_VERSION=5.2.2
+  - ES_VERSION=5.3.0
 
 os: linux
 

curator/cli.py

@@ -23,6 +23,7 @@
     'delete_snapshots' : DeleteSnapshots,
     'forcemerge' : ForceMerge,
     'open' : Open,
+    'reindex' : Reindex,
     'replicas' : Replicas,
     'restore' : Restore,
     'rollover' : Rollover,

curator/config_utils.py

@@ -1,5 +1,4 @@
 from voluptuous import Schema
-# from .defaults import settings
 from .validators import SchemaCheck, config_file
 from .utils import *
 from .logtools import LogInfo, Whitelist, Blacklist

curator/defaults/client_defaults.py

@@ -0,0 +1,42 @@
+from voluptuous import *
+
+# Configuration file: client
+def config_client():
+    return {
+        Optional('hosts', default='127.0.0.1'): Any(None, str, unicode, list),
+        Optional('port', default=9200): Any(
+            None, All(Coerce(int), Range(min=1, max=65535))
+        ),
+        Optional('url_prefix', default=''): Any(None, str, unicode),
+        Optional('use_ssl', default=False): Boolean(),
+        Optional('certificate', default=None): Any(None, str, unicode),
+        Optional('client_cert', default=None): Any(None, str, unicode),
+        Optional('client_key', default=None): Any(None, str, unicode),
+        Optional('aws_key', default=None): Any(None, str, unicode),
+        Optional('aws_secret_key', default=None): Any(None, str, unicode),
+        Optional('aws_region', default=None): Any(None, str, unicode),
+        Optional('ssl_no_validate', default=False): Boolean(),
+        Optional('http_auth', default=None): Any(None, str, unicode),
+        Optional('timeout', default=30): All(
+            Coerce(int), Range(min=1, max=86400)),
+        Optional('master_only', default=False): Boolean(),
+    }
+
+# Configuration file: logging
+def config_logging():
+    return {
+        Optional(
+            'loglevel', default='INFO'): Any(None,
+            'NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL',
+            All(Coerce(int), Any(0, 10, 20, 30, 40, 50))
+        ),
+        Optional('logfile', default=None): Any(None, str, unicode),
+        Optional(
+            'logformat', default='default'): Any(None, All(
+                Any(str, unicode),
+                Any('default', 'json', 'logstash')
+            )
+        ),
+        Optional(
+            'blacklist', default=['elasticsearch', 'urllib3']): Any(None, list),
+    }

curator/validators/filter_elements.py → curator/defaults/filter_elements.py

@@ -1,5 +1,5 @@
 from voluptuous import *
-from ..defaults import settings
+from . import settings
 
 ### Schema information ###
 

curator/validators/filtertypes.py → curator/defaults/filtertypes.py

@@ -1,5 +1,5 @@
 from voluptuous import *
-from ..defaults import settings
+from . import settings
 from . import filter_elements
 import logging
 logger = logging.getLogger(__name__)

curator/defaults/option_defaults.py

@@ -0,0 +1,253 @@
+from voluptuous import *
+
+# Action Options
+
+def allocation_type():
+    return { Optional('allocation_type', default='require'): All(
+        Any(str, unicode), Any('require', 'include', 'exclude')) }
+
+def conditions():
+    return {
+        Optional('conditions'): {
+            Optional('max_age'): Any(str, unicode),
+            Optional('max_docs'): Coerce(int)
+        }
+    }
+
+def continue_if_exception():
+    return { Optional('continue_if_exception', default=False): Boolean() }
+
+def count():
+    return { Required('count'): All(Coerce(int), Range(min=0, max=10)) }
+
+def delay():
+    return {
+        Optional('delay', default=0): All(
+                Coerce(float), Range(min=0.0, max=3600.0)
+            )
+    }
+
+def delete_aliases():
+    return { Optional('delete_aliases', default=False): Boolean() }
+
+def disable_action():
+    return { Optional('disable_action', default=False): Boolean() }
+
+def extra_settings():
+    return { Optional('extra_settings', default={}): dict }
+
+def ignore_empty_list():
+    return { Optional('ignore_empty_list', default=False): Boolean() }
+
+def ignore_unavailable():
+    return { Optional('ignore_unavailable', default=False): Boolean() }
+
+def include_aliases():
+    return { Optional('include_aliases', default=False): Boolean() }
+
+def include_global_state():
+    return { Optional('include_global_state', default=True): Boolean() }
+
+def indices():
+    return { Optional('indices', default=None): Any(None, list) }
+
+def key():
+    return { Required('key'): Any(str, unicode) }
+
+def max_num_segments():
+    return {
+        Required('max_num_segments'): All(Coerce(int), Range(min=1, max=32768))
+    }
+
+def max_wait(action):
+    # The separation is here in case I want to change defaults later...
+    value = -1
+    # if action in ['allocation', 'cluster_routing', 'replicas']:
+    #     value = -1
+    # elif action in ['restore', 'snapshot', 'reindex']:
+    #     value = -1
+    return { Optional('max_wait', default=value): Any(-1, Coerce(int), None) }
+
+def name(action):
+    if action in ['alias', 'create_index', 'rollover']:
+        return { Required('name'): Any(str, unicode) }
+    elif action == 'snapshot':
+        return {
+            Optional('name', default='curator-%Y%m%d%H%M%S'): Any(str, unicode)
+        }
+    elif action == 'restore':
+        return { Optional('name'): Any(str, unicode) }
+
+def partial():
+    return { Optional('partial', default=False): Boolean() }
+
+def refresh():
+    return { Optional('refresh', default=True): Boolean() }
+
+def remote_aws_key():
+    return { Optional('remote_aws_key', default=None): Any(str, unicode, None) }
+
+def remote_aws_secret_key():
+    return { Optional('remote_aws_secret_key', default=None): Any(str, unicode, None) }
+
+def remote_aws_region():
+    return { Optional('remote_aws_region', default=None): Any(str, unicode, None) }
+
+def remote_certificate():
+    return { Optional('remote_certificate', default=None): Any(str, unicode, None) }
+
+def remote_client_cert():
+    return { Optional('remote_client_cert', default=None): Any(str, unicode, None) }
+
+def remote_client_key():
+    return { Optional('remote_client_key', default=None): Any(str, unicode, None) }
+
+def remote_filters():
+    # This is really just a basic check here.  The real check is in the 
+    # validate_actions() method in utils.py
+    return { Optional('remote_filters', default=[
+                {
+                  'filtertype': 'pattern',
+                  'kind': 'regex',
+                  'value': '.*',
+                  'exclude': True,
+                }
+            ]
+        ): Any(list, None) 
+    }
+
+def remote_ssl_no_validate():
+    return { Optional('remote_ssl_no_validate', default=False): Boolean() }
+
+def remote_url_prefix():
+    return { Optional('remote_url_prefix', default=''): Any(None, str, unicode) }
+
+def rename_pattern():
+    return { Optional('rename_pattern'): Any(str, unicode) }
+
+def rename_replacement():
+    return { Optional('rename_replacement'): Any(str, unicode) }
+
+def repository():
+    return { Required('repository'): Any(str, unicode) }
+
+def request_body():
+    return {
+        Required('request_body'): {
+            Optional('conflicts'): Any(str, unicode),
+            Optional('size'): Coerce(int),
+            Required('source'): { 
+                Required('index'): Any(Any(str, unicode), list),
+                Optional('remote'): {
+                    Optional('host'): Any(str, unicode),
+                    Optional('headers'): Any(str, unicode),
+                    Optional('username'): Any(str, unicode),
+                    Optional('password'): Any(str, unicode),
+                    Optional('socket_timeout'): Any(str, unicode),
+                    Optional('connect_timeout'): Any(str, unicode),
+                },
+                Optional('type'): Any(Any(str, unicode), list),
+                Optional('query'): dict,
+                Optional('sort'): dict,
+                Optional('_source'): Any(Any(str, unicode), list),
+            },
+            Required('dest'): { 
+                Required('index'): Any(str, unicode),
+                Optional('type'): Any(Any(str, unicode), list),
+                Optional('op_type'): Any(str, unicode),
+                Optional('version_type'): Any(str, unicode),
+                Optional('routing'): Any(str, unicode),
+                Optional('pipeline'): Any(str, unicode),
+            },
+            Optional('script'): dict,
+        }
+    }
+
+def requests_per_second():
+    return { Optional('requests_per_second', default=-1): Any(
+            -1, Coerce(int), None)
+    }
+
+def retry_count():
+    return {
+        Optional('retry_count', default=3): All(
+                Coerce(int), Range(min=0, max=100)
+            )
+    }
+
+def retry_interval():
+    return {
+        Optional('retry_interval', default=120): All(
+                Coerce(int), Range(min=1, max=600)
+            )
+    }
+
+def routing_type():
+    return { Required('routing_type'): Any('allocation', 'rebalance') }
+
+def cluster_routing_setting():
+    return { Required('setting'): Any('enable') }
+
+def cluster_routing_value():
+    return {
+        Required('value'): Any(
+                'all', 'primaries', 'none', 'new_primaries', 'replicas'
+            )
+    }
+
+def skip_repo_fs_check():
+    return { Optional('skip_repo_fs_check', default=False): Boolean() }
+
+def slices():
+    return { Optional('slices', default=1): Any(
+        All(Coerce(int), Range(min=1, max=500)), None)
+    }
+
+def timeout(action):
+    # if action == 'reindex':
+    value = 60
+    return { Optional('timeout', default=value): Any(Coerce(int), None) }
+
+def timeout_override(action):
+    if action in ['forcemerge', 'restore', 'snapshot']:
+        value = 21600
+    elif action == 'close':
+        value = 180
+    else:
+        value = None
+
+    return {
+        Optional('timeout_override', default=value): Any(Coerce(int), None)
+    }
+
+def value():
+    return { Required('value', default=None): Any(str, unicode, None) }
+
+def wait_for_active_shards(action):
+    value = 0
+    if action == 'reindex':
+        value = 1
+    return {
+        Optional('wait_for_active_shards', default=value): Any(
+            Coerce(int), None)
+    }
+
+def wait_for_completion(action):
+    # if action in ['reindex', 'restore', 'snapshot']:
+    value = True
+    if action in ['allocation', 'cluster_routing', 'replicas']:
+        value = False
+    return { Optional('wait_for_completion', default=value): Boolean() }
+
+def wait_interval(action):
+    minval = 1
+    maxval = 30
+    # if action in ['allocation', 'cluster_routing', 'replicas']:
+    value = 3
+    if action in ['restore', 'snapshot', 'reindex']:
+        value = 9
+    return { Optional('wait_interval', default=value): Any(All(
+                Coerce(int), Range(min=minval, max=maxval)), None) }
+
+def warn_if_no_indices():
+    return { Optional('warn_if_no_indices', default=False): Boolean() }

curator/defaults/settings.py

@@ -48,6 +48,7 @@ def index_actions():
         'delete_indices',
         'forcemerge',
         'open',
+        'reindex',
         'replicas',
         'rollover',
         'snapshot',
@@ -90,3 +91,27 @@ def default_options():
 
 def default_filters():
     return { 'filters' : [{ 'filtertype' : 'none' }] }
+
+def structural_filter_elements():
+    return {
+        Optional('aliases'): Any(str, [str], unicode, [unicode]),
+        Optional('allocation_type'): Any(str, unicode),
+        Optional('count'): Coerce(int),
+        Optional('direction'): Any(str, unicode),
+        Optional('disk_space'): float,
+        Optional('epoch'): Any(Coerce(int), None),
+        Optional('exclude'): Any(int, str, unicode, bool, None),
+        Optional('field'): Any(str, unicode, None),
+        Optional('key'): Any(str, unicode),
+        Optional('kind'): Any(str, unicode),
+        Optional('max_num_segments'): Coerce(int),
+        Optional('reverse'): Any(int, str, unicode, bool, None),
+        Optional('source'): Any(str, unicode),
+        Optional('state'): Any(str, unicode),
+        Optional('stats_result'): Any(str, unicode, None),
+        Optional('timestring'): Any(str, unicode, None),
+        Optional('unit'): Any(str, unicode),
+        Optional('unit_count'): Coerce(int),
+        Optional('use_age'): Boolean(),
+        Optional('value'): Any(int, float, str, unicode, bool),
+    }

curator/exceptions.py

@@ -38,3 +38,8 @@ class SnapshotInProgress(ActionError):
     """
     Exception raised when a snapshot is already in progress
     """
+
+class ActionTimeout(CuratorException):
+    """
+    Exception raised when an action fails to complete in the allotted time
+    """

curator/indexlist.py

@@ -65,11 +65,11 @@ def __get_indices(self):
         self.loggit.debug('Getting all indices')
         self.all_indices = get_indices(self.client)
         self.indices = self.all_indices[:]
-        self.empty_list_check()
-        for index in self.indices:
-            self.__build_index_info(index)
-        self._get_metadata()
-        self._get_index_stats()
+        if self.indices:
+            for index in self.indices:
+                self.__build_index_info(index)
+            self._get_metadata()
+            self._get_index_stats()
 
     def __build_index_info(self, index):
         """