1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
|
#
# Copyright 2017 Huawei Technologies Co.,LTD.
#
# Licensed 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.
"""Command line tool for migrating metrics data from ceilometer native
storage backend to Gnocchi.
Usage:
python migrate_data_to_gnocchi.py --native-metering-connection "mysql+pymysql:
//root:password@127.0.0.1/ceilometer?charset=utf8"
NOTE:
You may need to install *tqdm* python lib to support progress bar in migration.
"""
import sys
try:
from tqdm import tqdm as progress_bar
except ImportError:
progress_bar = None
import argparse
from oslo_config import cfg
from oslo_db import options as db_options
from oslo_log import log
from ceilometer.dispatcher import gnocchi
from ceilometer import service
from ceilometer import storage
from ceilometer.storage import impl_mongodb
from ceilometer.storage import impl_sqlalchemy
from ceilometer.storage.mongo import utils as pymongo_utils
from ceilometer import utils
def get_parser():
parser = argparse.ArgumentParser(
description='For migrating metrics data from ceilometer built-in '
'storage backends to Gnocchi.')
parser.add_argument(
'--native-metering-connection',
required=True,
help='The database connection url of native storage backend. '
'e.g. mysql+pymysql://root:password@127.0.0.1/ceilometer?charset'
'=utf8',
)
parser.add_argument(
'--ceilometer-config-file',
help="The config file of ceilometer, it is main used for gnocchi "
"dispatcher to init gnocchiclient with the service credentials "
"defined in the ceilometer config file. Default as "
"/etc/ceilometer/ceilometer.conf",
)
parser.add_argument(
'--log-file',
default='gnocchi-data-migration.log',
help="The log file to record messages during migration.", )
parser.add_argument(
'--batch-migration-size',
default=300,
help="The amount of samples that will be posted to gnocchi per batch",)
parser.add_argument(
'--start-timestamp',
default=None,
help="The stat timestamp of metrics data to be migrated, with ISO8601 "
"format, e.g. 2016-01-25 11:58:00",
)
parser.add_argument(
'--end-timestamp',
default=None,
help="The end timestamp of metrics data to be migrated, with ISO8601 "
"format, e.g. 2017-01-25 11:58:00",
)
return parser
def count_samples(storage_conn, start_timestamp=None, end_timestamp=None):
if start_timestamp:
start_timestamp = utils.sanitize_timestamp(start_timestamp)
if start_timestamp:
end_timestamp = utils.sanitize_timestamp(end_timestamp)
if isinstance(storage_conn, impl_sqlalchemy.Connection):
from ceilometer.storage.sqlalchemy import models
session = storage_conn._engine_facade.get_session()
query = session.query(models.Sample.id)
if start_timestamp:
query = query.filter(models.Sample.timestamp >= start_timestamp)
if end_timestamp:
query = query.filter(models.Sample.timestamp < end_timestamp)
return query.count()
elif isinstance(storage_conn, impl_mongodb.Connection):
ts_range = pymongo_utils.make_timestamp_range(start_timestamp,
end_timestamp)
return storage_conn.db.meter.count(ts_range)
else:
print("Unsupported type of storage connection: %s" % storage_conn)
sys.exit(1)
def get_native_storage_conn(metering_connection):
storage_conf = cfg.ConfigOpts()
db_options.set_defaults(storage_conf)
storage_conf.register_opts(storage.OPTS, 'database')
storage_conf.set_override('metering_connection',
metering_connection, 'database')
storage_conn = storage.get_connection_from_config(storage_conf)
return storage_conn
def main():
args = get_parser().parse_args()
storage_conn = get_native_storage_conn(args.native_metering_connection)
total_amount = count_samples(storage_conn, args.start_timestamp,
args.end_timestamp)
print('%s samples will be migrated to Gnocchi.' % total_amount)
# NOTE: we need service credentials to init gnocchiclient
config_file = ([args.ceilometer_config_file] if
args.ceilometer_config_file else None)
gnocchi_conf = service.prepare_service([], config_file)
logger = log.getLogger()
log_conf = cfg.ConfigOpts()
log.register_options(log_conf)
log_conf.set_override('log_file', args.log_file)
log_conf.set_override('debug', True)
log.setup(log_conf, 'ceilometer_migration')
time_filters = []
if args.start_timestamp:
time_filters.append({">=": {'timestamp': args.start_timestamp}})
if args.end_timestamp:
time_filters.append({"<": {'timestamp': args.end_timestamp}})
gnocchi_dispatcher = gnocchi.GnocchiDispatcher(gnocchi_conf)
batch_size = args.batch_migration_size
if total_amount == 'Unknown':
total_amount = None
orderby = [{"message_id": "asc"}]
last_message_id = None
migrated_amount = 0
if progress_bar:
pbar = progress_bar(total=total_amount, ncols=100, unit='samples')
else:
pbar = None
while migrated_amount < total_amount:
if time_filters and last_message_id:
filter_expr = {
'and': time_filters + [{">": {"message_id": last_message_id}}]}
elif time_filters and not last_message_id:
if len(time_filters) == 1:
filter_expr = time_filters[0]
else:
filter_expr = {'and': time_filters}
elif not time_filters and last_message_id:
filter_expr = {">": {"message_id": last_message_id}}
else:
filter_expr = None
samples = storage_conn.query_samples(
filter_expr=filter_expr, orderby=orderby, limit=batch_size)
samples = list(samples)
if not samples:
break
last_message_id = samples[-1].message_id
for sample in samples:
logger.info('Migrating sample with message_id: %s, meter: %s, '
'resource_id: %s' % (sample.message_id,
sample.counter_name,
sample.resource_id))
samples_dict = [sample.as_dict() for sample in samples]
gnocchi_dispatcher.record_metering_data(samples_dict)
length = len(samples)
migrated_amount += length
if pbar:
pbar.update(length)
logger.info("=========== %s metrics data migration done ============" %
total_amount)
if __name__ == '__main__':
sys.exit(main())
|
