nginx+lua+mod_cache+mod_proxy+RANGE

bitmovr.lua

--
--	bitmovr.lua
--	simple lua webserver which starts to listen on a socket, and
--	forwards all GET calls it receives to a backend server
--	this is extremely lightweight, as it will move the bits from one
--	socket to another, without any disk i/o
--
--	Depends on md5, io, LuaSockets and Memcached.lua
--
--	Application flow:
--	1. File is requested: /m/xyz.jpg
--	2. A check is done if this file is already in the registers somewhere
--	   (eg:404, 5xx, etc.)
--	3. If the file was found in the registers, return the http code from the
--	   registers
--	4. If the file was not found, get the headers remotely, and strip some
--	   of the headers we don't want. We follow any redirects as well
--	5. If the response is non-valid, we set that response in the registers
--	   and return the correct http code (with some of the headers we got).
--	6. If the response is valid, do a proper GET request to the file and
--	   stream the contents directly to the client.
--	7. ---
--	8. Profit.
-- 
--	Created by Sander van de Graaf on 2010-06-28.
--	Copyright Sanoma Digital. All rights reserved.
--
 
-- parse the config file
local settings = {}
for line in io.lines('./bitmovr.conf') do
	for key,value in string.gmatch(line, '([a-zA-Z0-9\_\@\,\:\\\/\.]+) ?= ?\"?([a-zA-Z0-9\_\@\,\:\\\/\.]+)\"?') do
		if value == 'false' then
		 value = false
		end
		if value == 'true' then
		 value = true
		end
		settings[key] = value;
	end
end
settings.log = true;
 
-- include libraries
local md5 = require("md5")
local io = require("io");
local httpSocket = require("socket.http");
local ltn12 = require("ltn12");
require('Memcached');
 
-- this will hold any non-200 files
errors = {}
spinners = {}
missing = {}
found = {}
 
-- load namespace
local socket = require("socket")
 
-- fetch the hostname and port based on the arg, or let luaSocket decide
assignedHostArg = arg[1] or '0.0.0.0:0'
local assignedHost, assignedPort = string.match(assignedHostArg,'^([0-9\.]+):([0-9]+);?$');
 
-- create a TCP socket and bind it to the local host at the given port
server = assert(socket.bind(assignedHost, assignedPort))
 
-- find out which port the OS chose for us
ip, port = server:getsockname();
hostname = socket.dns.gethostname(ip);
 
-- log function for logging messages to disk or stdout
function logit(msg)
	if msg == nil then
		msg = 'EMPTY'
	end
		print('[' .. port .. '] ' .. msg);
		logger = io.open(settings.logfile, 'a+');
		logger:write('[' .. port .. '] ' .. msg .. "\n");
		logger:close();
end
 
-- get a memcached connection to multiple servers
function getMemcacheConnection()
	servers = {}
	fields = {}
	settings.memcached_servers:gsub("([^,]*)"..',', function(c) table.insert(fields, c) end)
	for i,descr in ipairs(fields) do
				   local server, port = string.match(descr,'^([0-9\.]+):([0-9]+)$');
				   table.insert(servers,{server,port})
	end
	return Memcached.Connect(servers);
end
 
-- update the stats in memcached, so we can see what's happening
function updateStats(type, what)
	-- type can be of: 200, 201, 400, 404, 504, 'total' and 'sent'
	local key = hostname .. port .. type;
 
	-- connect to local memcached host
	local memcache = getMemcacheConnection();
 
	-- increase or set the amount
	result = memcache:incr(key, what);
	if result == 'NOT_FOUND' then
				   memcache:set(key,what);
				   result = what
	end
	logit('increased ' .. key .. ' with ' .. what .. ': ' .. result);
			   
	memcache:disconnect_all()
end
 
logit("Bitmovr listening");
 
-- loop forever waiting for clients
while 1 do
	-- wait for a connection from any client
	local client = server:accept()
	-- make sure we don't block waiting for this client's line
	client:settimeout(10)
	-- receive the line
	local line, err = client:receive()

	-- if there was no error, send it back to the client
	if not err then
		logit(line);

		i = true
		requestHeaders = {};
		headerHash = '';
		while i == true do
			lastLine = client:receive();
			local key,value = string.match(lastLine,'^(.+): (.*)$');
			logit(lastLine)
			if lastLine ~= '' and string.lower(key) == 'range' then
				requestHeaders[key] = value;
				headerHash = key .. value
			else
				i = false
			end
		end
		
		for key, value in pairs(requestHeaders) do
			logit('R: ' .. key .. ': ' .. value .. '\n');
		end
		-- logit(requestHeaders)
		logit('done!')

		-- we do need a proper http 1.1 GET request
		-- we strip out any get args, we ignore those!
		local requestFilename = string.match(line,'^.+ (/m[\/a-zA-Z0-9_\.\-]+).* HTTP/1..$');
		local otherRequest = string.match(line,'^.+ \/([status|statistics]).*$');

		if requestFilename ~= nil and otherRequest == nil then

			-- unique name
			fileHash = md5.sumhexa(requestFilename)
			
			-- check if the hash is in memcached
			local memcache = getMemcacheConnection()
			hash = nil
			hash = memcache:get('hash_' .. fileHash)
			
			logit(hash)
			
			if hash == nil then
				
				-- get the headers for this file remotely
				local b = {};
				r, c, h = socket.http.request{
					method = 'GET',
					url = 'http://' .. settings.backend_host .. requestFilename,
					redirect = false,
					headers = requestHeaders
				}

				logit(c)
				
				if c == 302 then
					-- woohoo, this file exists, add it to memcached
					status = 'found';
					for header, value in pairs(h) do
						if string.lower(header) == 'location' then
							logit('found location: ' .. value);
							hash = string.match(value,'^http://.+/(.+)$');
						end
					end
		
					logit('found hash in location: ' .. hash);
					
					-- if the hash still is nil, then the location header was wrong :(
					if hash ~= nil then
						local memcache = getMemcacheConnection()
						memcache:set('hash_' .. fileHash, hash);
						memcache:disconnect_all()
					end
				else
					-- this is a spinner
					if c == 200 or c == 206 then

						-- send response code
						client:send('HTTP/1.1 ' .. c .. '\r\n');

						-- send out headers
						for header, value in pairs(h) do
							if string.lower(header) == 'content-length' or
								string.lower(header) == 'last-modified' or
								string.lower(header) == 'content-range' then
									client:send(header .. ': ' .. value .. '\r\n');
							end
						end
						
						-- send out newline according to RFC
						client:send('\r\n');
						
						local outputSink = socket.sink("close-when-done", client)
						r, c, h = socket.http.request{
							method = 'GET',
							url = 'http://' .. settings.backend_host .. requestFilename,
							redirect = false,
							headers = requestHeaders,
							sink = outputSink
						}

						done = true
					else
						-- send response code
						client:send('HTTP/1.1 ' .. c .. '\r\n');
						client:send('Connection: Close\r\n');
						client:send('\r\n');

						client:send('uhoh, woops!');
						done = true
					end
					
				end
			end
			
			if hash ~= nil then
				r, c, h = socket.http.request{
					method = 'GET',
					url = settings.binaryBackend .. hash,
					redirect = true,
					headers = requestHeaders
				}
					
				-- send response code
				client:send('HTTP/1.1 ' .. c .. '\r\n');
				
				-- send out headers
				for header, value in pairs(h) do
					if string.lower(header) == 'content-length' or
						string.lower(header) == 'last-modified' or
						string.lower(header) == 'content-range' then
							client:send(header .. ': ' .. value .. '\r\n');
					end
				end
				
				-- send out newline according to RFC
				client:send('\r\n');
					
				-- fetch the binary data
				local outputSink = socket.sink("close-when-done", client)
				local r, c, h = socket.http.request{
					sink = outputSink,
					method = 'GET',
					url = settings.binaryBackend .. hash,
					redirect = true,
					headers = requestHeaders
				}
				outputSink = nil;

			else
				if done == false or done == nil then
					logit('foobar1')
					client:send('HTTP/1.1 502 Upstream error :(\r\n');
				end
			end
		else
			logit('foobar2')
			client:send('HTTP/1.1 502 Upstream error, dunno what to do :(\r\n');
		end
		
		-- close the connection
		client:close()
	end
end

nginx.conf

		location ~ \.mp4 {
			# set a different cache header
			proxy_cache_key "$scheme$proxy_host$request_uri $http_range";
			proxy_set_header Range $http_range;

                        proxy_cache mediatoolrange;
                        proxy_cache_valid 1; # cache for 10 days
                        proxy_cache_valid 404 10; # cache for 10 seconds
                        proxy_cache_valid any 1m; # 502's etc for 1 minute
                        proxy_pass http://backend;
		}

		location /m {
                        proxy_cache mediatool;
                        proxy_cache_valid 1; # cache for 10 days
                        proxy_cache_valid 404 10; # cache for 10 seconds 
                        proxy_cache_valid any 1m; # 502's etc for 1 minute
                        proxy_pass http://backend;
		}