Changeset 48

Timestamp:

07/19/04 21:23:59 (4 years ago)

Author:

marksmith

Message:

Update MogileFS to support HTTP for all operations -- get, put, copy, and
delete. This is made to interact with a Perlbal setup using mogstored as
the frontend.

People wishing to use the old style MogileFS on a local filesystem or NFS
should turn USE_HTTP to 0 at the top of mogilefsd.

Files:

• trunk/api/perl/MogileFS.pm (modified) (6 diffs)
• trunk/devnotes/sql.txt (modified) (2 diffs)
• trunk/server/mogilefsd (modified) (11 diffs)
• trunk/server/mogstored (modified) (1 diff)

trunk/api/perl/MogileFS.pm

@@ -23 +23 @@
-use strict;
-use Carp;
+use IO::WrapTie;
-use fields qw(root domain backend);
-
@@ -50 +51 @@
-sub new_file {
-    my MogileFS $self = shift;
-
+, $bytes
-
-    my $res = $self->{backend}->do_request
@@ -60 +61 @@
-
-    # create a MogileFS::NewFile object, based off of IO::File
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-}
-
@@ -80 +93 @@
-        }) or return undef;
-
-
+
+
+
-}
-
@@ -124 +139 @@
-    # FIXME: add actual validation
-    {
-
-
-
+
+
+
+
+
+
+
-        $self->{domain} = $args{domain} or
-            _fail("constructor requires parameter 'domain'");
@@ -491 +510 @@
-}
-
+################################################################################
+# MogileFS::HTTPFile object
+# NOTE: This is meant to be used within IO::WrapTie...
+#
+
+package MogileFS::NewHTTPFile;
+
+use fields ('host',
+            'sock', # IO::Socket; created only when we need it
+            'uri',
+            'data', # buffered data we have
+            'pos', # simulated file position
+            'length', # length of data field
+            'content_length', # declared length of data we will be receiving (not required)
+            'mg',
+            'fid',
+            'devid',
+            'class',
+            'key',
+            'path', # full URL to save data to
+            );
+
+sub TIEHANDLE {
+    my MogileFS::NewHTTPFile $self = shift;
+    $self = fields::new($self) unless ref $self;
+
+    my %args = @_;
+    return undef unless $args{path} =~ m!http://(.+?)(/.+)$!;
+
+    $self->{host} = $1;
+    $self->{uri} = $2;
+    $self->{data} = '';
+    $self->{length} = 0;
+    $self->{content_length} = $args{content_length} + 0;
+    $self->{pos} = 0;
+    $self->{$_} = $args{$_} foreach qw(mg fid devid class key path);
+    
+    return $self;
+}
+*new = *TIEHANDLE;
+
+sub PRINT {
+    my MogileFS::NewHTTPFile $self = shift;
+
+    # get data to send to server
+    my $data = shift;
+    my $newlen = length $data;
+    $self->{pos} += $newlen;
+
+    # now make socket if we don't have one
+    if (!$self->{sock} && $self->{content_length}) {
+        $self->{sock} = IO::Socket::INET->new(PeerAddr => $self->{host})
+            or die "Error: unable to open socket: $!\n";
+        $self->{sock}->print("PUT $self->{uri} HTTP/1.0\r\nContent-length: $self->{content_length}\r\n\r\n");
+    }
+
+    # write some data to our socket
+    if ($self->{sock}) {
+        # store on data if we're under 1k
+        if ($self->{length} < 1024) {
+            if ($self->{length} + $newlen > 1024) {
+                $self->{length} = 1024;
+                $self->{data} .= substr($data, 0, 1024 - $self->{length});
+            } else {
+                $self->{length} += $newlen;
+                $self->{data} .= $data;
+            }
+        }
+
+        # actually write
+        $self->{sock}->print($data);
+    } else {
+        # or not, just stick it on our queued data
+        $self->{data} .= $data;
+        $self->{length} += $newlen;
+    }
+}
+*print = *PRINT;
+
+# get/set functions
+sub _getset {
+    my MogileFS::NewHTTPFile $self = shift;
+    my $what = shift;
+    
+    if (@_) {
+        # note: we're a TIEHANDLE interface, so we're not QUITE like a
+        # normal class... our parameters tend to come in via an arrayref
+        my $val = shift;
+        $val = shift(@$val) if ref $val eq 'ARRAY';
+        return $self->{$what} = $val;
+    } else {
+        return $self->{$what};
+    }
+}
+sub path  { _getset(shift, 'path');      }
+sub class { _getset(shift, 'class', @_); }
+sub key   { _getset(shift, 'key', @_);   }
+
+sub CLOSE {
+    my MogileFS::NewHTTPFile $self = shift;
+
+    # if we're closed and we have no sock...
+    unless ($self->{sock}) {
+        $self->{sock} = IO::Socket::INET->new(PeerAddr => $self->{host})
+            or die "Error: unable to open socket: $!\n";
+        $self->{sock}->print("PUT $self->{uri} HTTP/1.0\r\nContent-length: $self->{length}\r\n\r\n");
+        $self->{sock}->print($self->{data});
+    }
+    
+    # get response from put
+    if ($self->{sock}) {
+        my $line = $self->{sock}->getline;
+        if ($line =~ m!^HTTP/\d+\.\d+\s+(\d+)!) {
+            # all 2xx responses are success
+            unless ($1 >= 200 && $1 <= 299) {
+                $@ = "HTTP response $1 from upload\n";
+                return undef;
+            }
+        } else {
+            $@ = "Response line not understood: $line\n";
+            return undef;
+        }
+        $self->{sock}->close;
+    }
+
+    my MogileFS $mg = $self->{mg};
+    my $domain = $mg->{domain};
+
+    my $fid   = $self->{fid};
+    my $devid = $self->{devid};
+    my $path  = $self->{path};
+
+    my $key = shift || $self->{key};
+
+    $mg->{backend}->do_request
+        ("create_close", {
+            fid    => $fid,
+            devid  => $devid,
+            domain => $domain,
+            size   => $self->{content_length} ? $self->{content_length} : $self->{length},
+            key    => $key,
+            path   => $path,
+        }) or return undef;
+
+    return 1;
+}
+*close = *CLOSE;
+
+sub TELL {
+    # return our current pos
+    return $_[0]->{pos};
+}
+*tell = *TELL;
+
+sub SEEK {
+    # simply set pos...
+    die "ERROR: Seek past end of file\n" if $_[1] > $_[0]->{length};
+    $_[0]->{pos} = $_[1];
+}
+*seek = *SEEK;
+
+sub EOF {
+    return ($_[0]->{pos} >= $_[0]->{length}) ? 1 : 0;
+}
+*eof = *EOF;
+
+sub BINMODE {
+    # no-op, we're always in binary mode
+}
+*binmode = *BINMODE;
+
+sub READ {
+    my MogileFS::NewHTTPFile $self = shift;
+    my $count = $_[1] + 0;
+    
+    my $max = $self->{length} - $self->{pos};
+    $max = $count if $count < $max;
+
+    $_[0] = substr($self->{data}, $self->{pos}, $max);
+    $self->{pos} += $max;
+
+    return $max;
+}
+*read = *READ;
+
+sub AUTOLOAD {
+    use vars qw($AUTOLOAD);
+    warn "Error: $AUTOLOAD not implemented.\n";
+}
+
-1;

trunk/devnotes/sql.txt

@@ -27 +27 @@
-      classname     VARCHAR(50),
-      UNIQUE      (dmid,classname),
-,
+
-);
-
@@ -127 +127 @@
-
-   status     ENUM('alive','dead','down'),
+   http_port  MEDIUMINT UNSIGNED DEFAULT 7500,
-
-   hostname   VARCHAR(40),

trunk/server/mogilefsd

@@ -40 +40 @@
-$DEFAULT_MOG_ROOT = "/mnt/mogilefs";
-$DEBUG = 0;
+
+use constant USE_HTTP => 1;
-
-my (
@@ -320 +322 @@
-
-                my $path = make_path($devid, $fid);
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-                # device is timing out.  take note of it and
@@ -346 +380 @@
-
-    }
+}
+
+# copies a file from one Perlbal to another utilizing HTTP
+sub http_copy {
+    my ($sdevid, $ddevid, $fid) = @_;
+
+    # get some information we'll need
+    my $devs = Mgd::get_device_summary();
+    my ($sdev, $ddev) = ($devs->{$sdevid}, $devs->{$ddevid});
+    unless (ref $sdev && ref $ddev) {
+        warn "Error: unable to get device information: source=$sdevid, destination=$ddevid, fid=$fid\n";
+        return 0;
+    }
+    my ($spath, $dpath) = (Mgd::make_http_path($sdevid, $fid),
+                           Mgd::make_http_path($ddevid, $fid));
+    my ($shost, $sport) = (Mgd::hostid_ip($sdev->{hostid}), Mgd::hostid_http_port($sdev->{hostid}));
+    my ($dhost, $dport) = (Mgd::hostid_ip($ddev->{hostid}), Mgd::hostid_http_port($ddev->{hostid}));
+    unless (defined $spath && defined $dpath && defined $shost && defined $dhost && $sport && $dport) {
+        # show detailed information to find out what's not configured right
+        warn "Error: unable to replicate file fid=$fid from device id $sdevid to device id $ddevid\n";
+        warn "       http://$shost:$sport$spath -> http://$dhost:$dport$dpath\n";
+        return 0;
+    }
+    print "http://$shost:$sport$spath -> http://$dhost:$dport$dpath\n" if $Mgd::DEBUG >= 2;
+
+    # setup our pipe error handler, in case we get closed on
+    my $pipe_closed = 0;
+    local $SIG{PIPE} = sub { $pipe_closed = 1; };
+
+    # okay, now get the file
+    my $sock = IO::Socket::INET->new(PeerAddr => $shost, PeerPort => $sport, Timeout => 2)
+        or return 0;
+    print $sock "GET $spath HTTP/1.0\r\n\r\n";
+    return 0 if $pipe_closed;
+
+    # we just want a content length
+    my $clen;
+    while (defined (my $line = <$sock>)) {
+        $line =~ s/[\s\r\n]+$//;
+        last unless length $line;
+        if ($line =~ m!^HTTP/\d+\.\d+\s+(\d+)!) {
+            # make sure we get a good response
+            unless ($1 >= 200 && $1 <= 299) {
+                warn "Error: Resource http://$shost:$sport$spath failed: HTTP $1\n";
+                return 0;
+            }
+        }
+        next unless $line =~ /^Content-length:\s*(\d+)\s*$/i;
+        $clen = $1;
+    }
+    return 0 unless $clen;
+
+    # open target for put
+    my $dsock = IO::Socket::INET->new(PeerAddr => $dhost, PeerPort => $dport, Timeout => 2) or return 0;
+    $dsock->write("PUT $dpath HTTP/1.0\r\nContent-length: $clen\r\n\r\n") or return 0;
+    return 0 if $pipe_closed;
+
+    # now read data and print while we're reading
+    my ($data, $read) = ('', 0);
+    while (!$pipe_closed && (my $bytes = read($sock, $data, $clen - $read))) {
+        $read += $bytes;
+        last unless $dsock->write($data);
+    }
+
+    # now read in the response line (should be first line)
+    my $line = <$dsock>;
+    if ($line =~ m!^HTTP/\d+\.\d+\s+(\d+)!) {
+        return 1 if $1 >= 200 && $1 <= 299;
+        warn "Error: got a 404 in put: device not on host?: http://$dhost:$dport$dpath" if $1 == 404;
+    } else {
+        warn "Error: HTTP response line not recognized: $line";
+    }
+    return 0;
-}
-
@@ -408 +515 @@
-        $sdevid ||= @exist_devid[int(rand(scalar @exist_devid))];
-
-
-
-
+
+
+
+
+
+
+
+
-
-        return $retunlock->(0) unless $rv;
@@ -536 +648 @@
-    %cache_host = ();
-    my $dbh = get_dbh();
-
+http_port, 
-    $sth->execute;
-    $cache_host{$_->{hostid}} = $_ while $_ = $sth->fetchrow_hashref;
@@ -549 +661 @@
-}
-
+# get size of file, return 0 on error
+sub get_file_size {
+    my $path = shift;
+
+    # quick case -- just a file on disk
+    unless ($path =~ m!^http://(.+?)(/.+)$!) {
+        return -s "$Mgd::MOG_ROOT/$path"
+    }
+
+    # URL; use a HEAD request to get the size of the file
+    my $sock = IO::Socket::INET->new(PeerAddr => $1, Timeout => 2)
+        or return 0;
+    print $sock "HEAD $2 HTTP/1.0\r\n\r\n";
+    while (defined (my $line = <$sock>)) {
+        if ($line =~ /^Content-length: (\d+)/i) {
+            return $1+0;
+        }
+    }
+
+    # no content length found?
+    return 0;
+}
+
-sub domain_id {
-    my $domain = shift;
@@ -567 +702 @@
-}
-
+sub hostid_ip {
+    my $hostid = shift;
+    check_host_cache();
+    my $h = $cache_host{$hostid};
+    return $h ? $h->{hostip} : undef;
+}
+
+sub hostid_http_port {
+    my $hostid = shift;
+    check_host_cache();
+    my $h = $cache_host{$hostid};
+    return $h ? $h->{http_port} : undef;
+}
+
+sub make_http_path {
+    my ($devid, $fid) = @_;
+
+    my $dsum = get_device_summary();
+    my $dinfo = $dsum->{$devid};
+    return undef unless $dinfo;
+    my $hostname = hostid_name($dinfo->{hostid});
+
+    my $nfid = sprintf '%010d', $fid;
+    my ( $b, $mmm, $ttt, $hto ) = ( $nfid =~ m{(\d)(\d{3})(\d{3})(\d{3})} );
+
+    return "/dev$devid/$b/$mmm/$ttt/$nfid.fid";
+}
+
+sub make_full_url {
+    my ($devid, $fid) = @_;
+
+    # get some information we'll need
+    my $devs = Mgd::get_device_summary();
+    my $dev = $devs->{$devid} or return undef;
+    my $path = Mgd::make_http_path($devid, $fid) or return undef;
+    my $host = Mgd::hostid_ip($dev->{hostid}) or return undef;
+    my $port = Mgd::hostid_http_port($dev->{hostid}) or return undef;
+    return "http://$host:$port$path";
+}
+
+# if given an HTTP URL, break it down into [ host, port, URI ], else
+# returns undef
+sub is_url {
+    my $path = shift;
+    if ($path =~ m!^http://(.+?)(?::(\d+))?(/.+)$!) {
+        return [ $1, $2 || 80, $3 ];
+    }
+    return undef;
+}
+
-sub make_path {
+    # jump out if we should be using HTTP stuff
+    return Mgd::make_full_url(@_) if Mgd::USE_HTTP;
+    
-    my ($devid, $fid) = @_;
-
@@ -819 +1007 @@
-    }
-
-
+
+
+
+
-
-    # TODO: check for EIO?
@@ -915 +1106 @@
-
-    my $key = $args->{key};
+
-    return $self->err_line("no_key") unless length($key);
-
@@ -923 +1115 @@
-    my $dbh = Mgd::get_dbh or
-        return $self->err_line("nodb");
-
+
-    my $filerow = Mgd::key_filerow($dbh, $dmid, $key);
-    return $self->err_line("unknown_key") unless $filerow;
@@ -941 +1133 @@
-        next unless $dev && $dev->{status} eq "alive";
-        my $path = Mgd::make_path($devid, $fid);
--s "$Mgd::MOG_ROOT/$path"
+Mgd::get_file_size($path)
-        my $n = ++$ret->{paths};
-        $ret->{"path$n"} = $path;
-        last if $n == 2;   # one verified, one likely seems enough for now.  time will tell.
-    }
+
-    return $self->ok_line($ret);
-}

trunk/server/mogstored

@@ -25 +25 @@
-
-my $path = "/var/mogdata";
-809
+750
-
-my $conf = "