Mercurial > logstash

comparison logstash_index_cleaner.py @ 3:796ac0b50dbf

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
add cron.daily index cleaning
author Carl Byington <carl@five-ten-sg.com>
date Thu, 07 Mar 2013 10:41:01 -0800
parents
children
comparison
equal deleted inserted replaced
2:9e0cdf091b8a 3:796ac0b50dbf
1 #!/usr/bin/env python
2 #
3 # Deletes all indices with a datestamp older than "days-to-keep" for daily
4 # if you have hourly indices, it will delete all of those older than "hours-to-keep"
5 #
6 # This script presumes an index is named typically, e.g. logstash-YYYY.MM.DD
7 # It will work with any name-YYYY.MM.DD or name-YYYY.MM.DD.HH type sequence
8 #
9 # Requires python and the following dependencies (all pip/easy_installable):
10 #
11 # pyes (python elasticsearch bindings, which might need simplejson)
12 # argparse (built-in in python2.7 and higher, python 2.6 and lower will have to easy_install it)
13 #
14 # TODO: Proper logging instead of just print statements, being able to configure a decent logging level.
15 # Unit tests. The code is somewhat broken up into logical parts that may be tested separately.
16 # Better error reporting?
17 # Improve the get_index_epoch method to parse more date formats. Consider renaming (to "parse_date_to_timestamp"?)
18
19 import sys
20 import time
21 import argparse
22 from datetime import timedelta
23
24 import pyes
25
26
27 __version__ = '0.1.2'
28
29
30 def make_parser():
31 """ Creates an ArgumentParser to parse the command line options. """
32 parser = argparse.ArgumentParser(description='Delete old logstash indices from Elasticsearch.')
33
34 parser.add_argument('-v', '--version', action='version', version='%(prog)s '+__version__)
35
36 parser.add_argument('--host', help='Elasticsearch host.', default='localhost')
37 parser.add_argument('--port', help='Elasticsearch port', default=9200, type=int)
38 parser.add_argument('-t', '--timeout', help='Elasticsearch timeout', default=30, type=int)
39
40 parser.add_argument('-p', '--prefix', help='Prefix for the indices. Indices that do not have this prefix are skipped.', default='logstash-')
41 parser.add_argument('-s', '--separator', help='Time unit separator', default='.')
42
43 parser.add_argument('-H', '--hours-to-keep', action='store', help='Number of hours to keep.', type=int)
44 parser.add_argument('-d', '--days-to-keep', action='store', help='Number of days to keep.', type=int)
45 parser.add_argument('-g', '--disk-space-to-keep', action='store', help='Disk space to keep (GB).', type=float)
46
47 parser.add_argument('-n', '--dry-run', action='store_true', help='If true, does not perform any changes to the Elasticsearch indices.', default=False)
48
49 return parser
50
51
52 def get_index_epoch(index_timestamp, separator='.'):
53 """ Gets the epoch of the index.
54
55 :param index_timestamp: A string on the format YYYY.MM.DD[.HH]
56 :return The creation time (epoch) of the index.
57 """
58 year_month_day_optionalhour = index_timestamp.split(separator)
59 if len(year_month_day_optionalhour) == 3:
60 year_month_day_optionalhour.append('3')
61
62 return time.mktime([int(part) for part in year_month_day_optionalhour] + [0,0,0,0,0])
63
64
65 def find_expired_indices(connection, days_to_keep=None, hours_to_keep=None, separator='.', prefix='logstash-', out=sys.stdout, err=sys.stderr):
66 """ Generator that yields expired indices.
67
68 :return: Yields tuples on the format ``(index_name, expired_by)`` where index_name
69 is the name of the expired index and expired_by is the number of seconds (a float value) that the
70 index was expired by.
71 """
72 utc_now_time = time.time() + time.altzone
73 days_cutoff = utc_now_time - days_to_keep * 24 * 60 * 60 if days_to_keep is not None else None
74 hours_cutoff = utc_now_time - hours_to_keep * 60 * 60 if hours_to_keep is not None else None
75
76 for index_name in sorted(set(connection.get_indices().keys())):
77 if not index_name.startswith(prefix):
78 print >> out, 'Skipping index due to missing prefix {0}: {1}'.format(prefix, index_name)
79 continue
80
81 unprefixed_index_name = index_name[len(prefix):]
82
83 # find the timestamp parts (i.e ['2011', '01', '05'] from '2011.01.05') using the configured separator
84 parts = unprefixed_index_name.split(separator)
85
86 # perform some basic validation
87 if len(parts) < 3 or len(parts) > 4 or not all([item.isdigit() for item in parts]):
88 print >> err, 'Could not find a valid timestamp from the index: {0}'.format(index_name)
89 continue
90
91 # find the cutoff. if we have more than 3 parts in the timestamp, the timestamp includes the hours and we
92 # should compare it to the hours_cutoff, otherwise, we should use the days_cutoff
93 cutoff = hours_cutoff
94 if len(parts) == 3:
95 cutoff = days_cutoff
96
97 # but the cutoff might be none, if the current index only has three parts (year.month.day) and we're only
98 # removing hourly indices:
99 if cutoff is None:
100 print >> out, 'Skipping {0} because it is of a type (hourly or daily) that I\'m not asked to delete.'.format(index_name)
101 continue
102
103 index_epoch = get_index_epoch(unprefixed_index_name)
104
105 # if the index is older than the cutoff
106 if index_epoch < cutoff:
107 yield index_name, cutoff-index_epoch
108
109 else:
110 print >> out, '{0} is {1} above the cutoff.'.format(index_name, timedelta(seconds=index_epoch-cutoff))
111
112 def find_overusage_indices(connection, disk_space_to_keep, separator='.', prefix='logstash-', out=sys.stdout, err=sys.stderr):
113 """ Generator that yields over usage indices.
114
115 :return: Yields tuples on the format ``(index_name, 0)`` where index_name
116 is the name of the expired index. The second element is only here for
117 compatiblity reasons.
118 """
119
120 disk_usage = 0.0
121 disk_limit = disk_space_to_keep * 2**30
122
123 for index_name in reversed(sorted(set(connection.get_indices().keys()))):
124
125 if not index_name.startswith(prefix):
126 print >> out, 'Skipping index due to missing prefix {0}: {1}'.format(prefix, index_name)
127 continue
128
129 index_size = connection.status(index_name).get('indices').get(index_name).get('index').get('primary_size_in_bytes')
130 disk_usage += index_size
131
132 if disk_usage > disk_limit:
133 yield index_name, 0
134 else:
135 print >> out, 'keeping {0}, disk usage is {1:.3f} GB and disk limit is {2:.3f} GB.'.format(index_name, disk_usage/2**30, disk_limit/2**30)
136
137 def main():
138 start = time.time()
139
140 parser = make_parser()
141 arguments = parser.parse_args()
142
143 if not arguments.hours_to_keep and not arguments.days_to_keep and not arguments.disk_space_to_keep:
144 print >> sys.stderr, 'Invalid arguments: You must specify either the number of hours, the number of days to keep or the maximum disk space to use'
145 parser.print_help()
146 return
147
148 connection = pyes.ES('{0}:{1}'.format(arguments.host, arguments.port), timeout=arguments.timeout)
149
150 if arguments.days_to_keep:
151 print 'Deleting daily indices older than {0} days.'.format(arguments.days_to_keep)
152 expired_indices = find_expired_indices(connection, arguments.days_to_keep, arguments.hours_to_keep, arguments.separator, arguments.prefix)
153 if arguments.hours_to_keep:
154 print 'Deleting hourly indices older than {0} hours.'.format(arguments.hours_to_keep)
155 expired_indices = find_expired_indices(connection, arguments.days_to_keep, arguments.hours_to_keep, arguments.separator, arguments.prefix)
156 if arguments.disk_space_to_keep:
157 print 'Let\'s keep disk usage lower than {} GB.'.format(arguments.disk_space_to_keep)
158 expired_indices = find_overusage_indices(connection, arguments.disk_space_to_keep, arguments.separator, arguments.prefix)
159
160 print ''
161
162 for index_name, expired_by in expired_indices:
163 expiration = timedelta(seconds=expired_by)
164
165 if arguments.dry_run:
166 print 'Would have attempted deleting index {0} because it is {1} older than the calculated cutoff.'.format(index_name, expiration)
167 continue
168
169 print 'Deleting index {0} because it was {1} older than cutoff.'.format(index_name, expiration)
170
171 deletion = connection.delete_index_if_exists(index_name)
172 # ES returns a dict on the format {u'acknowledged': True, u'ok': True} on success.
173 if deletion.get('ok'):
174 print 'Successfully deleted index: {0}'.format(index_name)
175 else:
176 print 'Error deleting index: {0}. ({1})'.format(index_name, deletion)
177
178 print ''
179 print 'Done in {0}.'.format(timedelta(seconds=time.time()-start))
180
181
182 if __name__ == '__main__':
183 main()