Line data Source code
1 : #include "utils/fd_ssarchive.h"
2 : #include "utils/fd_ssctrl.h"
3 : #include "utils/fd_sshttp.h"
4 :
5 : #include "../../disco/topo/fd_topo.h"
6 : #include "../../disco/metrics/fd_metrics.h"
7 :
8 : #include <errno.h>
9 : #include <fcntl.h>
10 : #include <unistd.h>
11 : #include <sys/socket.h>
12 :
13 : #include "generated/fd_snapld_tile_seccomp.h"
14 :
15 : #define NAME "snapld"
16 :
17 : /* The snapld tile is responsible for loading data from the local file
18 : or from an HTTP/TCP connection and sending it to the snapdc tile
19 : for later decompression. */
20 :
21 : typedef struct fd_snapld_tile {
22 :
23 : struct {
24 : char path[ PATH_MAX ];
25 : } config;
26 :
27 : int state;
28 : int load_full;
29 : int load_file;
30 : int sent_meta;
31 :
32 : int local_full_fd;
33 : int local_incr_fd;
34 :
35 : fd_sshttp_t * sshttp;
36 :
37 : struct {
38 : void const * base;
39 : } in_rd;
40 :
41 : struct {
42 : fd_wksp_t * mem;
43 : ulong chunk0;
44 : ulong wmark;
45 : ulong chunk;
46 : ulong mtu;
47 : } out_dc;
48 :
49 : } fd_snapld_tile_t;
50 :
51 : static ulong
52 0 : scratch_align( void ) {
53 0 : return fd_ulong_max( alignof(fd_snapld_tile_t), fd_sshttp_align() );
54 0 : }
55 :
56 : static ulong
57 0 : scratch_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
58 0 : ulong l = FD_LAYOUT_INIT;
59 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
60 0 : l = FD_LAYOUT_APPEND( l, fd_sshttp_align(), fd_sshttp_footprint() );
61 0 : return FD_LAYOUT_FINI( l, scratch_align() );
62 0 : }
63 :
64 : static void
65 : privileged_init( fd_topo_t * topo,
66 0 : fd_topo_tile_t * tile ) {
67 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
68 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
69 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
70 :
71 0 : ulong full_slot = ULONG_MAX;
72 0 : ulong incr_slot = ULONG_MAX;
73 0 : char full_path[ PATH_MAX ] = { 0 };
74 0 : char incr_path[ PATH_MAX ] = { 0 };
75 0 : if( FD_UNLIKELY( -1==fd_ssarchive_latest_pair( tile->snapld.snapshots_path, 1,
76 0 : &full_slot, &incr_slot,
77 0 : full_path, incr_path ) ) ) {
78 0 : ctx->local_full_fd = -1;
79 0 : ctx->local_incr_fd = -1;
80 0 : } else {
81 0 : FD_TEST( full_slot!=ULONG_MAX );
82 :
83 0 : ctx->local_full_fd = open( full_path, O_RDONLY|O_CLOEXEC|O_NONBLOCK );
84 0 : if( FD_UNLIKELY( -1==ctx->local_full_fd ) ) FD_LOG_ERR(( "open() failed `%s` (%i-%s)", full_path, errno, fd_io_strerror( errno ) ));
85 :
86 0 : if( FD_LIKELY( incr_slot!=ULONG_MAX ) ) {
87 0 : ctx->local_incr_fd = open( incr_path, O_RDONLY|O_CLOEXEC|O_NONBLOCK );
88 0 : if( FD_UNLIKELY( -1==ctx->local_incr_fd ) ) FD_LOG_ERR(( "open() failed `%s` (%i-%s)", incr_path, errno, fd_io_strerror( errno ) ));
89 0 : }
90 0 : }
91 0 : }
92 :
93 : static ulong
94 : populate_allowed_fds( fd_topo_t const * topo,
95 : fd_topo_tile_t const * tile,
96 : ulong out_fds_cnt,
97 0 : int * out_fds ) {
98 0 : if( FD_UNLIKELY( out_fds_cnt<4UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
99 :
100 0 : ulong out_cnt = 0;
101 0 : out_fds[ out_cnt++ ] = 2UL; /* stderr */
102 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) ) {
103 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd();
104 0 : }
105 :
106 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
107 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
108 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
109 0 : if( FD_LIKELY( -1!=ctx->local_full_fd ) ) out_fds[ out_cnt++ ] = ctx->local_full_fd;
110 0 : if( FD_LIKELY( -1!=ctx->local_incr_fd ) ) out_fds[ out_cnt++ ] = ctx->local_incr_fd;
111 :
112 0 : return out_cnt;
113 0 : }
114 :
115 : static ulong
116 : populate_allowed_seccomp( fd_topo_t const * topo,
117 : fd_topo_tile_t const * tile,
118 : ulong out_cnt,
119 0 : struct sock_filter * out ) {
120 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
121 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
122 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
123 :
124 0 : populate_sock_filter_policy_fd_snapld_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->local_full_fd, (uint)ctx->local_incr_fd );
125 0 : return sock_filter_policy_fd_snapld_tile_instr_cnt;
126 0 : }
127 :
128 : static void
129 : unprivileged_init( fd_topo_t * topo,
130 0 : fd_topo_tile_t * tile ) {
131 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
132 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
133 0 : fd_snapld_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapld_tile_t), sizeof(fd_snapld_tile_t) );
134 0 : void * _sshttp = FD_SCRATCH_ALLOC_APPEND( l, fd_sshttp_align(), fd_sshttp_footprint() );
135 :
136 0 : fd_memcpy( ctx->config.path, tile->snapld.snapshots_path, PATH_MAX );
137 :
138 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
139 :
140 0 : ctx->sshttp = fd_sshttp_join( fd_sshttp_new( _sshttp ) );
141 0 : FD_TEST( ctx->sshttp );
142 :
143 0 : FD_TEST( tile->in_cnt==1UL );
144 0 : fd_topo_link_t const * in_link = &topo->links[ tile->in_link_id[ 0 ] ];
145 0 : FD_TEST( 0==strcmp( in_link->name, "snapct_ld" ) );
146 0 : ctx->in_rd.base = fd_topo_obj_wksp_base( topo, in_link->dcache_obj_id );
147 :
148 0 : FD_TEST( tile->out_cnt==1UL );
149 0 : fd_topo_link_t const * out_link = &topo->links[ tile->out_link_id[ 0 ] ];
150 0 : FD_TEST( 0==strcmp( out_link->name, "snapld_dc" ) );
151 0 : ctx->out_dc.mem = fd_topo_obj_wksp_base( topo, out_link->dcache_obj_id );
152 0 : ctx->out_dc.chunk0 = fd_dcache_compact_chunk0( ctx->out_dc.mem, out_link->dcache );
153 0 : ctx->out_dc.wmark = fd_dcache_compact_wmark ( ctx->out_dc.mem, out_link->dcache, out_link->mtu );
154 0 : ctx->out_dc.chunk = ctx->out_dc.chunk0;
155 0 : ctx->out_dc.mtu = out_link->mtu;
156 0 : }
157 :
158 : static int
159 0 : should_shutdown( fd_snapld_tile_t * ctx ) {
160 0 : return ctx->state==FD_SNAPSHOT_STATE_SHUTDOWN;
161 0 : }
162 :
163 : static void
164 0 : metrics_write( fd_snapld_tile_t * ctx ) {
165 0 : FD_MGAUGE_SET( SNAPLD, STATE, (ulong)(ctx->state) );
166 0 : }
167 :
168 : static void
169 : after_credit( fd_snapld_tile_t * ctx,
170 : fd_stem_context_t * stem,
171 : int * opt_poll_in FD_PARAM_UNUSED,
172 0 : int * charge_busy ) {
173 0 : if( ctx->state!=FD_SNAPSHOT_STATE_PROCESSING ) return;
174 :
175 0 : uchar * out = fd_chunk_to_laddr( ctx->out_dc.mem, ctx->out_dc.chunk );
176 :
177 0 : if( ctx->load_file ) {
178 0 : long result = read( ctx->load_full ? ctx->local_full_fd : ctx->local_incr_fd, out, ctx->out_dc.mtu );
179 0 : if( FD_UNLIKELY( result<=0L ) ) {
180 0 : if( result==0L ) ctx->state = FD_SNAPSHOT_STATE_FINISHING;
181 0 : else if( FD_UNLIKELY( errno!=EAGAIN && errno!=EINTR ) ) {
182 0 : FD_LOG_WARNING(( "read() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
183 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
184 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
185 0 : }
186 0 : } else {
187 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_DATA, ctx->out_dc.chunk, (ulong)result, 0UL, 0UL, 0UL );
188 0 : ctx->out_dc.chunk = fd_dcache_compact_next( ctx->out_dc.chunk, (ulong)result, ctx->out_dc.chunk0, ctx->out_dc.wmark );
189 0 : *charge_busy = 1;
190 0 : }
191 0 : } else {
192 0 : ulong data_len = ctx->out_dc.mtu;
193 0 : int result = fd_sshttp_advance( ctx->sshttp, &data_len, out, fd_log_wallclock() );
194 0 : switch( result ) {
195 0 : case FD_SSHTTP_ADVANCE_AGAIN:
196 0 : break;
197 0 : case FD_SSHTTP_ADVANCE_DATA: {
198 0 : if( FD_UNLIKELY( !ctx->sent_meta ) ) {
199 : /* On the first DATA return, the HTTP headers are available
200 : for use. We need to send this metadata downstream, but
201 : need to do so before any data frags. So, we copy any data
202 : we received with the headers (if any) to the next dcache
203 : chunk and then publish both in order. */
204 0 : ctx->sent_meta = 1;
205 0 : fd_ssctrl_meta_t * meta = (fd_ssctrl_meta_t *)out;
206 0 : ulong next_chunk = fd_dcache_compact_next( ctx->out_dc.chunk, sizeof(fd_ssctrl_meta_t), ctx->out_dc.chunk0, ctx->out_dc.wmark );
207 0 : memmove( fd_chunk_to_laddr( ctx->out_dc.mem, next_chunk ), out, data_len );
208 0 : char const * full_name;
209 0 : char const * incr_name;
210 0 : fd_sshttp_snapshot_names( ctx->sshttp, &full_name, &incr_name );
211 0 : meta->total_sz = fd_sshttp_content_len( ctx->sshttp );
212 0 : FD_TEST( meta->total_sz!=ULONG_MAX );
213 0 : fd_memcpy( meta->name, ctx->load_full ? full_name : incr_name, PATH_MAX );
214 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_META, ctx->out_dc.chunk, sizeof(fd_ssctrl_meta_t), 0UL, 0UL, 0UL );
215 0 : ctx->out_dc.chunk = next_chunk;
216 0 : }
217 0 : if( FD_LIKELY( data_len!=0UL ) ) {
218 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_DATA, ctx->out_dc.chunk, data_len, 0UL, 0UL, 0UL );
219 0 : ctx->out_dc.chunk = fd_dcache_compact_next( ctx->out_dc.chunk, data_len, ctx->out_dc.chunk0, ctx->out_dc.wmark );
220 0 : }
221 0 : *charge_busy = 1;
222 0 : break;
223 0 : }
224 0 : case FD_SSHTTP_ADVANCE_DONE:
225 0 : ctx->state = FD_SNAPSHOT_STATE_FINISHING;
226 0 : break;
227 0 : case FD_SSHTTP_ADVANCE_ERROR:
228 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
229 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
230 0 : break;
231 0 : default: FD_LOG_ERR(( "unexpected fd_sshttp_advance result %d", result ));
232 0 : }
233 0 : }
234 0 : }
235 :
236 : static int
237 : returnable_frag( fd_snapld_tile_t * ctx,
238 : ulong in_idx FD_PARAM_UNUSED,
239 : ulong seq FD_PARAM_UNUSED,
240 : ulong sig,
241 : ulong chunk,
242 : ulong sz,
243 : ulong ctl FD_PARAM_UNUSED,
244 : ulong tsorig FD_PARAM_UNUSED,
245 : ulong tspub FD_PARAM_UNUSED,
246 0 : fd_stem_context_t * stem ) {
247 0 : switch( sig ) {
248 :
249 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_FULL:
250 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_INCR: {
251 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
252 0 : FD_TEST( sz==sizeof(fd_ssctrl_init_t) );
253 0 : fd_ssctrl_init_t const * msg = fd_chunk_to_laddr_const( ctx->in_rd.base, chunk );
254 0 : ctx->load_full = sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL;
255 0 : ctx->load_file = msg->file;
256 0 : ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
257 0 : ctx->sent_meta = 0;
258 0 : if( ctx->load_file ) {
259 0 : if( FD_UNLIKELY( 0!=lseek( ctx->load_full ? ctx->local_full_fd : ctx->local_incr_fd, 0, SEEK_SET ) ) )
260 0 : FD_LOG_ERR(( "lseek(0) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
261 0 : } else {
262 0 : if( ctx->load_full ) fd_sshttp_init( ctx->sshttp, msg->addr, "/snapshot.tar.bz2", 17UL, fd_log_wallclock() );
263 0 : else fd_sshttp_init( ctx->sshttp, msg->addr, "/incremental-snapshot.tar.bz2", 29UL, fd_log_wallclock() );
264 0 : }
265 0 : break;
266 0 : }
267 :
268 0 : case FD_SNAPSHOT_MSG_CTRL_FAIL:
269 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
270 0 : ctx->state==FD_SNAPSHOT_STATE_FINISHING ||
271 0 : ctx->state==FD_SNAPSHOT_STATE_ERROR );
272 0 : fd_sshttp_cancel( ctx->sshttp );
273 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
274 0 : break;
275 :
276 0 : case FD_SNAPSHOT_MSG_CTRL_NEXT:
277 0 : case FD_SNAPSHOT_MSG_CTRL_DONE:
278 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
279 0 : ctx->state==FD_SNAPSHOT_STATE_FINISHING ||
280 0 : ctx->state==FD_SNAPSHOT_STATE_ERROR );
281 0 : if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_FINISHING ) ) {
282 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
283 0 : fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
284 0 : return 0;
285 0 : }
286 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
287 0 : break;
288 :
289 0 : case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN:
290 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
291 0 : ctx->state = FD_SNAPSHOT_STATE_SHUTDOWN;
292 0 : metrics_write( ctx ); /* ensures that shutdown state is written to metrics workspace before the tile actually shuts down */
293 0 : break;
294 :
295 : /* FD_SNAPSHOT_MSG_CTRL_ERROR and FD_SNAPSHOT_MSG_DATA are not possible */
296 0 : default: FD_LOG_ERR(( "invalid sig %lu", sig ));
297 0 : }
298 :
299 : /* Forward the control message down the pipeline */
300 0 : fd_stem_publish( stem, 0UL, sig, 0UL, 0UL, 0UL, 0UL, 0UL );
301 :
302 0 : return 0;
303 0 : }
304 :
305 : /* Up to one frag from after_credit plus one from returnable_frag */
306 0 : #define STEM_BURST 2UL
307 :
308 0 : #define STEM_LAZY 1000L
309 :
310 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapld_tile_t
311 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapld_tile_t)
312 :
313 : #define STEM_CALLBACK_SHOULD_SHUTDOWN should_shutdown
314 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
315 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
316 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
317 :
318 : #include "../../disco/stem/fd_stem.c"
319 :
320 : fd_topo_run_tile_t fd_tile_snapld = {
321 : .name = NAME,
322 : .populate_allowed_seccomp = populate_allowed_seccomp,
323 : .populate_allowed_fds = populate_allowed_fds,
324 : .scratch_align = scratch_align,
325 : .scratch_footprint = scratch_footprint,
326 : .privileged_init = privileged_init,
327 : .unprivileged_init = unprivileged_init,
328 : .run = stem_run,
329 : .keep_host_networking = 1,
330 : .allow_connect = 1,
331 : .rlimit_file_cnt = 5UL, /* stderr, log, http, full/incr local files */
332 : };
333 :
334 : #undef NAME
|
