Contains the Concourse pipeline definition for building a line-server container
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

210 lines
4.9 KiB

  1. package s3
  2. import (
  3. "io"
  4. "io/ioutil"
  5. "os"
  6. "strconv"
  7. "time"
  8. "github.com/andreimarcu/linx-server/backends"
  9. "github.com/andreimarcu/linx-server/helpers"
  10. "github.com/aws/aws-sdk-go/aws"
  11. "github.com/aws/aws-sdk-go/aws/awserr"
  12. "github.com/aws/aws-sdk-go/aws/session"
  13. "github.com/aws/aws-sdk-go/service/s3"
  14. "github.com/aws/aws-sdk-go/service/s3/s3manager"
  15. )
  16. type S3Backend struct {
  17. bucket string
  18. svc *s3.S3
  19. }
  20. func (b S3Backend) Delete(key string) error {
  21. _, err := b.svc.DeleteObject(&s3.DeleteObjectInput{
  22. Bucket: aws.String(b.bucket),
  23. Key: aws.String(key),
  24. })
  25. if err != nil {
  26. return err
  27. }
  28. return nil
  29. }
  30. func (b S3Backend) Exists(key string) (bool, error) {
  31. _, err := b.svc.HeadObject(&s3.HeadObjectInput{
  32. Bucket: aws.String(b.bucket),
  33. Key: aws.String(key),
  34. })
  35. return err == nil, err
  36. }
  37. func (b S3Backend) Head(key string) (metadata backends.Metadata, err error) {
  38. var result *s3.HeadObjectOutput
  39. result, err = b.svc.HeadObject(&s3.HeadObjectInput{
  40. Bucket: aws.String(b.bucket),
  41. Key: aws.String(key),
  42. })
  43. if err != nil {
  44. if aerr, ok := err.(awserr.Error); ok {
  45. if aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound" {
  46. err = backends.NotFoundErr
  47. }
  48. }
  49. return
  50. }
  51. metadata, err = unmapMetadata(result.Metadata)
  52. return
  53. }
  54. func (b S3Backend) Get(key string) (metadata backends.Metadata, r io.ReadCloser, err error) {
  55. var result *s3.GetObjectOutput
  56. result, err = b.svc.GetObject(&s3.GetObjectInput{
  57. Bucket: aws.String(b.bucket),
  58. Key: aws.String(key),
  59. })
  60. if err != nil {
  61. if aerr, ok := err.(awserr.Error); ok {
  62. if aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound" {
  63. err = backends.NotFoundErr
  64. }
  65. }
  66. return
  67. }
  68. metadata, err = unmapMetadata(result.Metadata)
  69. r = result.Body
  70. return
  71. }
  72. func mapMetadata(m backends.Metadata) map[string]*string {
  73. return map[string]*string{
  74. "Expiry": aws.String(strconv.FormatInt(m.Expiry.Unix(), 10)),
  75. "Delete_key": aws.String(m.DeleteKey),
  76. "Size": aws.String(strconv.FormatInt(m.Size, 10)),
  77. "Mimetype": aws.String(m.Mimetype),
  78. "Sha256sum": aws.String(m.Sha256sum),
  79. }
  80. }
  81. func unmapMetadata(input map[string]*string) (m backends.Metadata, err error) {
  82. expiry, err := strconv.ParseInt(aws.StringValue(input["Expiry"]), 10, 64)
  83. if err != nil {
  84. return m, err
  85. }
  86. m.Expiry = time.Unix(expiry, 0)
  87. m.Size, err = strconv.ParseInt(aws.StringValue(input["Size"]), 10, 64)
  88. if err != nil {
  89. return
  90. }
  91. m.DeleteKey = aws.StringValue(input["Delete_key"])
  92. m.Mimetype = aws.StringValue(input["Mimetype"])
  93. m.Sha256sum = aws.StringValue(input["Sha256sum"])
  94. return
  95. }
  96. func (b S3Backend) Put(key string, r io.Reader, expiry time.Time, deleteKey string) (m backends.Metadata, err error) {
  97. tmpDst, err := ioutil.TempFile("", "linx-server-upload")
  98. if err != nil {
  99. return m, err
  100. }
  101. defer tmpDst.Close()
  102. defer os.Remove(tmpDst.Name())
  103. bytes, err := io.Copy(tmpDst, r)
  104. if bytes == 0 {
  105. return m, backends.FileEmptyError
  106. } else if err != nil {
  107. return m, err
  108. }
  109. m, err = helpers.GenerateMetadata(r)
  110. if err != nil {
  111. return
  112. }
  113. m.Expiry = expiry
  114. m.DeleteKey = deleteKey
  115. // XXX: we may not be able to write this to AWS easily
  116. //m.ArchiveFiles, _ = helpers.ListArchiveFiles(m.Mimetype, m.Size, tmpDst)
  117. uploader := s3manager.NewUploaderWithClient(b.svc)
  118. input := &s3manager.UploadInput{
  119. Bucket: aws.String(b.bucket),
  120. Key: aws.String(key),
  121. Body: tmpDst,
  122. Metadata: mapMetadata(m),
  123. }
  124. _, err = uploader.Upload(input)
  125. if err != nil {
  126. return
  127. }
  128. return
  129. }
  130. func (b S3Backend) PutMetadata(key string, m backends.Metadata) (err error) {
  131. _, err = b.svc.CopyObject(&s3.CopyObjectInput{
  132. Bucket: aws.String(b.bucket),
  133. Key: aws.String(key),
  134. CopySource: aws.String("/" + b.bucket + "/" + key),
  135. Metadata: mapMetadata(m),
  136. MetadataDirective: aws.String("REPLACE"),
  137. })
  138. if err != nil {
  139. return
  140. }
  141. return
  142. }
  143. func (b S3Backend) Size(key string) (int64, error) {
  144. input := &s3.HeadObjectInput{
  145. Bucket: aws.String(b.bucket),
  146. Key: aws.String(key),
  147. }
  148. result, err := b.svc.HeadObject(input)
  149. if err != nil {
  150. return 0, err
  151. }
  152. return *result.ContentLength, nil
  153. }
  154. func (b S3Backend) List() ([]string, error) {
  155. var output []string
  156. input := &s3.ListObjectsInput{
  157. Bucket: aws.String(b.bucket),
  158. }
  159. results, err := b.svc.ListObjects(input)
  160. if err != nil {
  161. return nil, err
  162. }
  163. for _, object := range results.Contents {
  164. output = append(output, *object.Key)
  165. }
  166. return output, nil
  167. }
  168. func NewS3Backend(bucket string, region string, endpoint string, forcePathStyle bool) S3Backend {
  169. awsConfig := &aws.Config{}
  170. if region != "" {
  171. awsConfig.Region = aws.String(region)
  172. }
  173. if endpoint != "" {
  174. awsConfig.Endpoint = aws.String(endpoint)
  175. }
  176. if forcePathStyle == true {
  177. awsConfig.S3ForcePathStyle = aws.Bool(true)
  178. }
  179. sess := session.Must(session.NewSession(awsConfig))
  180. svc := s3.New(sess)
  181. return S3Backend{bucket: bucket, svc: svc}
  182. }