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
|
# 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.
import logging
import socket
import tempfile
from oslo_config import cfg
from glance_store import exceptions
from glance_store.i18n import _
LOG = logging.getLogger(__name__)
READ_SIZE = 65536
BUFFERING_OPTS = [
cfg.StrOpt('swift_upload_buffer_dir',
help="""
Directory to buffer image segments before upload to Swift.
Provide a string value representing the absolute path to the
directory on the glance node where image segments will be
buffered briefly before they are uploaded to swift.
NOTES:
* This is required only when the configuration option
``swift_buffer_on_upload`` is set to True.
* This directory should be provisioned keeping in mind the
``swift_store_large_object_chunk_size`` and the maximum
number of images that could be uploaded simultaneously by
a given glance node.
Possible values:
* String value representing an absolute directory path
Related options:
* swift_buffer_on_upload
* swift_store_large_object_chunk_size
"""),
]
CONF = cfg.CONF
def validate_buffering(buffer_dir):
if buffer_dir is None:
msg = _('Configuration option "swift_upload_buffer_dir" is '
'not set. Please set it to a valid path to buffer '
'during Swift uploads.')
raise exceptions.BadStoreConfiguration(store_name='swift',
reason=msg)
# NOTE(dharinic): Ensure that the provided directory path for
# buffering is valid
try:
_tmpfile = tempfile.TemporaryFile(dir=buffer_dir)
except OSError as err:
msg = _('Unable to use buffer directory set with '
'"swift_upload_buffer_dir". Error: %s') % err
raise exceptions.BadStoreConfiguration(store_name='swift',
reason=msg)
else:
_tmpfile.close()
return True
class BufferedReader(object):
"""Buffer a chunk (segment) worth of data to disk before sending it swift.
This creates the ability to back the input stream up and re-try put object
requests. (Swiftclient will try to reset the file pointer on any upload
failure if seek and tell methods are provided on the input file.)
Chunks are temporarily buffered to disk. Disk space consumed will be
roughly (segment size * number of in-flight upload requests).
There exists a possibility where the disk space consumed for buffering MAY
eat into the disk space available for glance cache. This may affect image
download performance. So, extra care should be taken while deploying this
to ensure there is enough disk space available.
"""
def __init__(self, fd, checksum, os_hash_value, total, verifier=None,
backend_group=None):
self.fd = fd
self.total = total
self.checksum = checksum
self.os_hash_value = os_hash_value
self.verifier = verifier
self.backend_group = backend_group
# maintain a pointer to use to update checksum and verifier
self.update_position = 0
if self.backend_group:
buffer_dir = getattr(CONF,
self.backend_group).swift_upload_buffer_dir
else:
buffer_dir = CONF.glance_store.swift_upload_buffer_dir
self._tmpfile = tempfile.TemporaryFile(dir=buffer_dir)
self._buffered = False
self.is_zero_size = False
self._buffer()
# Setting the file pointer back to the beginning of file
self._tmpfile.seek(0)
def read(self, size):
"""Read up to a chunk's worth of data from the input stream into a
file buffer. Then return data out of that buffer.
"""
remaining = self.total - self._tmpfile.tell()
read_size = min(remaining, size)
# read out of the buffered chunk
result = self._tmpfile.read(read_size)
# update the checksum and verifier with only the bytes
# they have not seen
update = self.update_position - self._tmpfile.tell()
if update < 0:
self.checksum.update(result[update:])
self.os_hash_value.update(result[update:])
if self.verifier:
self.verifier.update(result[update:])
self.update_position += abs(update)
return result
def _buffer(self):
to_buffer = self.total
LOG.debug("Buffering %s bytes of image segment" % to_buffer)
buffer_read_count = 0
while not self._buffered:
read_size = min(to_buffer, READ_SIZE)
try:
buf = self.fd.read(read_size)
except IOError as e:
# We actually don't know what exactly self.fd is. And as a
# result we don't know which exception it may raise. To pass
# the retry mechanism inside swift client we must limit the
# possible set of errors.
raise socket.error(*e.args)
if len(buf) == 0:
if self._tmpfile.tell() == 0:
self.is_zero_size = True
self._tmpfile.seek(0)
self._buffered = True
break
self._tmpfile.write(buf)
to_buffer -= len(buf)
buffer_read_count = buffer_read_count + 1
if buffer_read_count == 0:
self.is_zero_size = True
# NOTE(belliott) seek and tell get used by python-swiftclient to "reset"
# if there is a put_object error
def seek(self, offset):
LOG.debug("Seek from %s to %s" % (self._tmpfile.tell(), offset))
self._tmpfile.seek(offset)
def tell(self):
return self._tmpfile.tell()
@property
def bytes_read(self):
return self.tell()
def __enter__(self):
self._tmpfile.__enter__()
return self
def __exit__(self, type, value, traceback):
# close and delete the temporary file used to buffer data
self._tmpfile.__exit__(type, value, traceback)
|
