1 /*-
2 * Copyright (c) 2011, 2012, 2013, 2014 Spectra Logic Corporation
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
7 * are met:
8 * 1. Redistributions of source code must retain the above copyright
9 * notice, this list of conditions, and the following disclaimer,
10 * without modification.
11 * 2. Redistributions in binary form must reproduce at minimum a disclaimer
12 * substantially similar to the "NO WARRANTY" disclaimer below
13 * ("Disclaimer") and any redistribution must be conditioned upon
14 * including a substantially similar Disclaimer requirement for further
15 * binary redistribution.
16 *
17 * NO WARRANTY
18 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTIBILITY AND FITNESS FOR
21 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22 * HOLDERS OR CONTRIBUTORS BE LIABLE FOR SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
24 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
25 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
26 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING
27 * IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
28 * POSSIBILITY OF SUCH DAMAGES.
29 *
30 * Authors: Justin T. Gibbs (Spectra Logic Corporation)
31 */
32
33 /**
34 * \file consumer.cc
35 */
36
37 #include <sys/cdefs.h>
38 #include <sys/poll.h>
39 #include <sys/socket.h>
40 #include <sys/un.h>
41
42 #include <err.h>
43 #include <errno.h>
44 #include <fcntl.h>
45 #include <syslog.h>
46 #include <unistd.h>
47
48 #include <cstdarg>
49 #include <cstring>
50 #include <list>
51 #include <map>
52 #include <string>
53
54 #include "guid.h"
55 #include "event.h"
56 #include "event_factory.h"
57 #include "exception.h"
58
59 #include "consumer.h"
60
61 __FBSDID("$FreeBSD$");
62
63 /*================================== Macros ==================================*/
64 #define NUM_ELEMENTS(x) (sizeof(x) / sizeof(*x))
65
66 /*============================ Namespace Control =============================*/
67 using std::string;
68 namespace DevdCtl
69 {
70
71 /*============================= Class Definitions ============================*/
72 /*----------------------------- DevdCtl::Consumer ----------------------------*/
73 //- Consumer Static Private Data -----------------------------------------------
74 const char Consumer::s_devdSockPath[] = "/var/run/devd.seqpacket.pipe";
75
76 //- Consumer Public Methods ----------------------------------------------------
Consumer(Event::BuildMethod * defBuilder,EventFactory::Record * regEntries,size_t numEntries)77 Consumer::Consumer(Event::BuildMethod *defBuilder,
78 EventFactory::Record *regEntries,
79 size_t numEntries)
80 : m_devdSockFD(-1),
81 m_eventFactory(defBuilder),
82 m_replayingEvents(false)
83 {
84 m_eventFactory.UpdateRegistry(regEntries, numEntries);
85 }
86
~Consumer()87 Consumer::~Consumer()
88 {
89 DisconnectFromDevd();
90 }
91
92 bool
ConnectToDevd()93 Consumer::ConnectToDevd()
94 {
95 struct sockaddr_un devdAddr;
96 int sLen;
97 int result;
98
99 if (m_devdSockFD != -1) {
100 /* Already connected. */
101 syslog(LOG_DEBUG, "%s: Already connected.", __func__);
102 return (true);
103 }
104 syslog(LOG_INFO, "%s: Connecting to devd.", __func__);
105
106 memset(&devdAddr, 0, sizeof(devdAddr));
107 devdAddr.sun_family= AF_UNIX;
108 strlcpy(devdAddr.sun_path, s_devdSockPath, sizeof(devdAddr.sun_path));
109 sLen = SUN_LEN(&devdAddr);
110
111 m_devdSockFD = socket(AF_UNIX, SOCK_SEQPACKET | SOCK_NONBLOCK, 0);
112 if (m_devdSockFD == -1)
113 err(1, "Unable to create socket");
114 result = connect(m_devdSockFD,
115 reinterpret_cast<sockaddr *>(&devdAddr),
116 sLen);
117 if (result == -1) {
118 syslog(LOG_INFO, "Unable to connect to devd");
119 DisconnectFromDevd();
120 return (false);
121 }
122
123 syslog(LOG_INFO, "Connection to devd successful");
124 return (true);
125 }
126
127 void
DisconnectFromDevd()128 Consumer::DisconnectFromDevd()
129 {
130 if (m_devdSockFD != -1) {
131 syslog(LOG_INFO, "Disconnecting from devd.");
132 close(m_devdSockFD);
133 }
134 m_devdSockFD = -1;
135 }
136
137 std::string
ReadEvent()138 Consumer::ReadEvent()
139 {
140 char buf[MAX_EVENT_SIZE + 1];
141 ssize_t len;
142
143 len = ::recv(m_devdSockFD, buf, MAX_EVENT_SIZE, MSG_WAITALL);
144 if (len == -1)
145 return (std::string(""));
146 else {
147 /* NULL-terminate the result */
148 buf[len] = '\0';
149 return (std::string(buf));
150 }
151 }
152
153 void
ReplayUnconsumedEvents(bool discardUnconsumed)154 Consumer::ReplayUnconsumedEvents(bool discardUnconsumed)
155 {
156 EventList::iterator event(m_unconsumedEvents.begin());
157 bool replayed_any = (event != m_unconsumedEvents.end());
158
159 m_replayingEvents = true;
160 if (replayed_any)
161 syslog(LOG_INFO, "Started replaying unconsumed events");
162 while (event != m_unconsumedEvents.end()) {
163 bool consumed((*event)->Process());
164 if (consumed || discardUnconsumed) {
165 delete *event;
166 event = m_unconsumedEvents.erase(event);
167 } else {
168 event++;
169 }
170 }
171 if (replayed_any)
172 syslog(LOG_INFO, "Finished replaying unconsumed events");
173 m_replayingEvents = false;
174 }
175
176 bool
SaveEvent(const Event & event)177 Consumer::SaveEvent(const Event &event)
178 {
179 if (m_replayingEvents)
180 return (false);
181 m_unconsumedEvents.push_back(event.DeepCopy());
182 return (true);
183 }
184
185 Event *
NextEvent()186 Consumer::NextEvent()
187 {
188 if (!Connected())
189 return(NULL);
190
191 Event *event(NULL);
192 try {
193 string evString;
194
195 evString = ReadEvent();
196 if (! evString.empty()) {
197 Event::TimestampEventString(evString);
198 event = Event::CreateEvent(m_eventFactory, evString);
199 }
200 } catch (const Exception &exp) {
201 exp.Log();
202 DisconnectFromDevd();
203 }
204 return (event);
205 }
206
207 /* Capture and process buffered events. */
208 void
ProcessEvents()209 Consumer::ProcessEvents()
210 {
211 Event *event;
212 while ((event = NextEvent()) != NULL) {
213 if (event->Process())
214 SaveEvent(*event);
215 delete event;
216 }
217 }
218
219 void
FlushEvents()220 Consumer::FlushEvents()
221 {
222 std::string s;
223
224 do
225 s = ReadEvent();
226 while (! s.empty()) ;
227 }
228
229 bool
EventsPending()230 Consumer::EventsPending()
231 {
232 struct pollfd fds[1];
233 int result;
234
235 do {
236 fds->fd = m_devdSockFD;
237 fds->events = POLLIN;
238 fds->revents = 0;
239 result = poll(fds, NUM_ELEMENTS(fds), /*timeout*/0);
240 } while (result == -1 && errno == EINTR);
241
242 if (result == -1)
243 err(1, "Polling for devd events failed");
244
245 if ((fds->revents & POLLERR) != 0)
246 throw Exception("Consumer::EventsPending(): "
247 "POLLERR detected on devd socket.");
248
249 if ((fds->revents & POLLHUP) != 0)
250 throw Exception("Consumer::EventsPending(): "
251 "POLLHUP detected on devd socket.");
252
253 return ((fds->revents & POLLIN) != 0);
254 }
255
256 } // namespace DevdCtl
257