From: Ben Pfaff Date: Mon, 23 Nov 2009 21:59:20 +0000 (-0800) Subject: netflow: Break out code for sending packets into a new "collectors" module. X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=6bab37989b1b5e8981bee80e1fedef621799332c;p=openvswitch netflow: Break out code for sending packets into a new "collectors" module. sFlow uses a similar UDP structure so this will allow use to reuse the NetFlow code for it. --- diff --git a/lib/vlog-modules.def b/lib/vlog-modules.def index ce298b55..0d44e734 100644 --- a/lib/vlog-modules.def +++ b/lib/vlog-modules.def @@ -21,6 +21,7 @@ VLOG_MODULE(bridge) VLOG_MODULE(chain) VLOG_MODULE(cfg) VLOG_MODULE(cfg_mod) +VLOG_MODULE(collectors) VLOG_MODULE(controller) VLOG_MODULE(coverage) VLOG_MODULE(ctlpath) diff --git a/ofproto/automake.mk b/ofproto/automake.mk index 232d45f6..87a0fa68 100644 --- a/ofproto/automake.mk +++ b/ofproto/automake.mk @@ -7,6 +7,8 @@ noinst_LIBRARIES += ofproto/libofproto.a ofproto_libofproto_a_SOURCES = \ + ofproto/collectors.c \ + ofproto/collectors.h \ ofproto/discovery.c \ ofproto/discovery.h \ ofproto/executer.c \ diff --git a/ofproto/collectors.c b/ofproto/collectors.c new file mode 100644 index 00000000..f7cb1dbe --- /dev/null +++ b/ofproto/collectors.c @@ -0,0 +1,123 @@ +/* + * Copyright (c) 2008, 2009 Nicira Networks. + * + * 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. + */ + +#include + +#include "collectors.h" + +#include +#include +#include + +#include "socket-util.h" +#include "svec.h" +#include "util.h" + +#define THIS_MODULE VLM_collectors +#include "vlog.h" + +struct collectors { + int *fds; /* Sockets. */ + size_t n_fds; /* Number of sockets. */ +}; + +/* Opens the targets specified in 'targets' for sending UDP packets. This is + * useful for e.g. sending NetFlow or sFlow packets. Returns 0 if successful, + * otherwise a positive errno value if opening at least one collector failed. + * + * Each target in 'targets' should be a string in the format "[:]". + * may be omitted if 'default_port' is nonzero, in which case it + * defaults to 'default_port'. + * + * '*collectorsp' is set to a null pointer if no targets were successfully + * added, otherwise to a new collectors object if at least one was successfully + * added. Thus, even on a failure return, it is possible that '*collectorsp' + * is nonnull, and even on a successful return, it is possible that + * '*collectorsp' is null, if 'target's is an empty svec. */ +int +collectors_create(const struct svec *targets_, uint16_t default_port, + struct collectors **collectorsp) +{ + struct collectors *c; + struct svec targets; + int retval = 0; + size_t i; + + svec_clone(&targets, targets_); + svec_sort_unique(&targets); + + c = xmalloc(sizeof *c); + c->fds = xmalloc(sizeof *c->fds * targets.n); + c->n_fds = 0; + for (i = 0; i < targets.n; i++) { + const char *name = targets.names[i]; + int error; + int fd; + + error = inet_open_active(SOCK_DGRAM, name, default_port, NULL, &fd); + if (fd >= 0) { + c->fds[c->n_fds++] = fd; + } else { + static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5); + + VLOG_WARN_RL(&rl, "couldn't open connection to collector %s (%s)", + name, strerror(error)); + if (!retval) { + retval = error; + } + } + } + svec_destroy(&targets); + + if (c->n_fds) { + *collectorsp = c; + } else { + collectors_destroy(c); + *collectorsp = NULL; + } + + return retval; +} + +/* Destroys 'c'. */ +void +collectors_destroy(struct collectors *c) +{ + if (c) { + size_t i; + + for (i = 0; i < c->n_fds; i++) { + close(c->fds[i]); + } + free(c->fds); + free(c); + } +} + +/* Sends the 'n'-byte 'payload' to each of the collectors in 'c'. */ +void +collectors_send(const struct collectors *c, const void *payload, size_t n) +{ + size_t i; + + for (i = 0; i < c->n_fds; i++) { + static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5); + if (send(c->fds[i], payload, n, 0) == -1) { + VLOG_WARN_RL(&rl, "sending to collector failed: %s", + strerror(errno)); + } + } +} diff --git a/ofproto/collectors.h b/ofproto/collectors.h new file mode 100644 index 00000000..a4abb631 --- /dev/null +++ b/ofproto/collectors.h @@ -0,0 +1,31 @@ +/* + * Copyright (c) 2009 Nicira Networks. + * + * 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. + */ + +#ifndef COLLECTORS_H +#define COLLECTORS_H 1 + +#include +#include "svec.h" + +struct collectors; + +int collectors_create(const struct svec *targets, uint16_t default_port, + struct collectors **); +void collectors_destroy(struct collectors *); + +void collectors_send(const struct collectors *, const void *, size_t); + +#endif /* collectors.h */ diff --git a/ofproto/netflow.c b/ofproto/netflow.c index 6e09ad60..7c77c64f 100644 --- a/ofproto/netflow.c +++ b/ofproto/netflow.c @@ -21,6 +21,7 @@ #include #include #include "cfg.h" +#include "collectors.h" #include "flow.h" #include "netflow.h" #include "ofpbuf.h" @@ -95,8 +96,7 @@ struct netflow { uint8_t engine_type; /* Value of engine_type to use. */ uint8_t engine_id; /* Value of engine_id to use. */ long long int boot_time; /* Time when netflow_create() was called. */ - int *fds; /* Sockets for NetFlow collectors. */ - size_t n_fds; /* Number of Netflow collectors. */ + struct collectors *collectors; /* NetFlow collectors. */ bool add_id_to_iface; /* Put the 7 least signficiant bits of * 'engine_id' into the most signficant * bits of the interface fields. */ @@ -106,17 +106,6 @@ struct netflow { long long int reconfig_time; /* When we reconfigured the timeouts. */ }; -static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5); - -static int -open_collector(char *dst) -{ - int error, fd; - - error = inet_open_active(SOCK_DGRAM, dst, 0, NULL, &fd); - return fd >= 0 ? fd : -error; -} - void netflow_expire(struct netflow *nf, struct netflow_flow *nf_flow, struct ofexpired *expired) @@ -201,70 +190,25 @@ netflow_expire(struct netflow *nf, struct netflow_flow *nf_flow, void netflow_run(struct netflow *nf) { - size_t i; - - if (!nf->packet.size) { - return; + if (nf->packet.size) { + collectors_send(nf->collectors, nf->packet.data, nf->packet.size); + nf->packet.size = 0; } - - for (i = 0; i < nf->n_fds; i++) { - if (send(nf->fds[i], nf->packet.data, nf->packet.size, 0) == -1) { - VLOG_WARN_RL(&rl, "netflow message send failed: %s", - strerror(errno)); - } - } - nf->packet.size = 0; -} - -static void -clear_collectors(struct netflow *nf) -{ - size_t i; - - for (i = 0; i < nf->n_fds; i++) { - close(nf->fds[i]); - } - free(nf->fds); - nf->fds = NULL; - nf->n_fds = 0; } int netflow_set_options(struct netflow *nf, const struct netflow_options *nf_options) { - struct svec collectors; int error = 0; - size_t i; long long int old_timeout; nf->engine_type = nf_options->engine_type; nf->engine_id = nf_options->engine_id; nf->add_id_to_iface = nf_options->add_id_to_iface; - clear_collectors(nf); - - svec_clone(&collectors, &nf_options->collectors); - svec_sort_unique(&collectors); - - nf->fds = xmalloc(sizeof *nf->fds * collectors.n); - for (i = 0; i < collectors.n; i++) { - const char *name = collectors.names[i]; - char *tmpname = xstrdup(name); - int fd = open_collector(tmpname); - free(tmpname); - if (fd >= 0) { - nf->fds[nf->n_fds++] = fd; - } else { - VLOG_WARN("couldn't open connection to collector (%s), " - "ignoring %s\n", strerror(-fd), name); - if (!error) { - error = -fd; - } - } - } - - svec_destroy(&collectors); + collectors_destroy(nf->collectors); + collectors_create(&nf_options->collectors, 0, &nf->collectors); old_timeout = nf->active_timeout; if (nf_options->active_timeout != -1) { @@ -287,8 +231,7 @@ netflow_create(void) nf->engine_type = 0; nf->engine_id = 0; nf->boot_time = time_msec(); - nf->fds = NULL; - nf->n_fds = 0; + nf->collectors = NULL; nf->add_id_to_iface = false; nf->netflow_cnt = 0; ofpbuf_init(&nf->packet, 1500); @@ -300,7 +243,7 @@ netflow_destroy(struct netflow *nf) { if (nf) { ofpbuf_uninit(&nf->packet); - clear_collectors(nf); + collectors_destroy(nf->collectors); free(nf); } }